accidental.cc
podcast manager
refactoring the context a bit
+1
-1
eslint.config.js
+1
-1
eslint.config.js
···
88
// mostly cribbed from preact's config, but that's not setup to handle eslint9
89
// https://github.com/preactjs/eslint-config-preact/blob/master/index.js
90
name: 'client files',
91
+
files: ['src/client/**/*.@(js|ts)', 'src/**/*.@(jsx|tsx)'],
92
languageOptions: {
93
globals: {
94
...globals.es2024,
+2
-2
src/client/components/feed-import-nytimes.tsx
+2
-2
src/client/components/feed-import-nytimes.tsx
···
1
import {useSignal} from '@preact/signals'
2
import {useCallback} from 'preact/hooks'
3
4
-
import {useSkypod} from '#client/skypod/context'
5
import {useRealmIdentity} from '#realm/client/context-identity'
6
7
// NY Times RSS feeds from https://www.nytimes.com/rss
···
73
74
export const FeedImportNYTimes: preact.FunctionComponent = () => {
75
const {identity} = useRealmIdentity()
76
-
const store = useSkypod()
77
78
const importing$ = useSignal(false)
79
const imported$ = useSignal(0)
···
1
import {useSignal} from '@preact/signals'
2
import {useCallback} from 'preact/hooks'
3
4
+
import {useActionDispatch} from '#client/skypod/action-dispatch/context.js'
5
import {useRealmIdentity} from '#realm/client/context-identity'
6
7
// NY Times RSS feeds from https://www.nytimes.com/rss
···
73
74
export const FeedImportNYTimes: preact.FunctionComponent = () => {
75
const {identity} = useRealmIdentity()
76
+
const store = useActionDispatch()
77
78
const importing$ = useSignal(false)
79
const imported$ = useSignal(0)
+2
-2
src/client/components/feed-import-podcasts.tsx
+2
-2
src/client/components/feed-import-podcasts.tsx
···
1
import {useSignal} from '@preact/signals'
2
import {useCallback} from 'preact/hooks'
3
4
-
import {useSkypod} from '#client/skypod/context'
5
import {useRealmIdentity} from '#realm/client/context-identity'
6
7
// Popular podcasts
···
21
22
export const FeedImportPodcasts: preact.FunctionComponent = () => {
23
const {identity} = useRealmIdentity()
24
-
const store = useSkypod()
25
26
const importing$ = useSignal(false)
27
const imported$ = useSignal(0)
···
1
import {useSignal} from '@preact/signals'
2
import {useCallback} from 'preact/hooks'
3
4
+
import {useActionDispatch} from '#client/skypod/action-dispatch/context.js'
5
import {useRealmIdentity} from '#realm/client/context-identity'
6
7
// Popular podcasts
···
21
22
export const FeedImportPodcasts: preact.FunctionComponent = () => {
23
const {identity} = useRealmIdentity()
24
+
const store = useActionDispatch()
25
26
const importing$ = useSignal(false)
27
const imported$ = useSignal(0)
+2
-2
src/client/components/feed-import-tech.tsx
+2
-2
src/client/components/feed-import-tech.tsx
···
1
import {useSignal} from '@preact/signals'
2
import {useCallback} from 'preact/hooks'
3
4
-
import {useSkypod} from '#client/skypod/context'
5
import {useRealmIdentity} from '#realm/client/context-identity'
6
7
// Popular tech/programming feeds
···
20
21
export const FeedImportTech: preact.FunctionComponent = () => {
22
const {identity} = useRealmIdentity()
23
-
const store = useSkypod()
24
25
const importing$ = useSignal(false)
26
const imported$ = useSignal(0)
···
1
import {useSignal} from '@preact/signals'
2
import {useCallback} from 'preact/hooks'
3
4
+
import {useActionDispatch} from '#client/skypod/action-dispatch/context.js'
5
import {useRealmIdentity} from '#realm/client/context-identity'
6
7
// Popular tech/programming feeds
···
20
21
export const FeedImportTech: preact.FunctionComponent = () => {
22
const {identity} = useRealmIdentity()
23
+
const store = useActionDispatch()
24
25
const importing$ = useSignal(false)
26
const imported$ = useSignal(0)
+2
-2
src/client/components/messenger.tsx
+2
-2
src/client/components/messenger.tsx
···
2
import {useCallback} from 'preact/hooks'
3
4
import {useDatabase} from '#client/root/context-database'
5
-
import {useSkypod} from '#client/skypod/context'
6
import {useRealmIdentity} from '#realm/client/context-identity'
7
8
export const Messenger: preact.FunctionComponent = () => {
9
const {useDbSignal} = useDatabase()
10
const {identity} = useRealmIdentity()
11
-
const store = useSkypod()
12
13
const feeds$ = useDbSignal((db) => db.feeds.toArray())
14
···
2
import {useCallback} from 'preact/hooks'
3
4
import {useDatabase} from '#client/root/context-database'
5
+
import {useActionDispatch} from '#client/skypod/action-dispatch/context'
6
import {useRealmIdentity} from '#realm/client/context-identity'
7
8
export const Messenger: preact.FunctionComponent = () => {
9
const {useDbSignal} = useDatabase()
10
const {identity} = useRealmIdentity()
11
+
const store = useActionDispatch()
12
13
const feeds$ = useDbSignal((db) => db.feeds.toArray())
14
+15
-21
src/client/page-app.tsx
+15
-21
src/client/page-app.tsx
···
1
-
import {DatabaseProvider} from '#client/root/context-database'
2
import {SkypodProvider} from '#client/skypod/context'
3
import {RealmConnectionManager} from '#realm/client/components/connection-manager'
4
-
import {
5
-
RealmConnectionFallbackProps,
6
-
RealmConnectionProvider,
7
-
} from '#realm/client/context-connection'
8
-
import {RealmIdentityFallbackProps, RealmIdentityProvider} from '#realm/client/context-identity'
9
10
import {DebugNuke} from './components/debug-nuke'
11
import {FeedImportNYTimes} from './components/feed-import-nytimes'
···
26
const wsurl = `${wsproto}://${wshost}/stream`
27
28
return (
29
-
<DatabaseProvider>
30
-
<RealmIdentityProvider fallback={identityFallback}>
31
-
<RealmConnectionProvider fallback={connectionFallback} url={wsurl}>
32
-
<SkypodProvider>
33
-
<RealmConnectionManager />
34
-
<PeerList />
35
-
<FeedImportNYTimes />
36
-
<FeedImportTech />
37
-
<FeedImportPodcasts />
38
-
<Messenger />
39
-
<DebugNuke />
40
-
</SkypodProvider>
41
-
</RealmConnectionProvider>
42
-
</RealmIdentityProvider>
43
-
</DatabaseProvider>
44
)
45
}
···
1
import {SkypodProvider} from '#client/skypod/context'
2
import {RealmConnectionManager} from '#realm/client/components/connection-manager'
3
+
import {RealmConnectionFallbackProps} from '#realm/client/context-connection'
4
+
import {RealmIdentityFallbackProps} from '#realm/client/context-identity'
5
6
import {DebugNuke} from './components/debug-nuke'
7
import {FeedImportNYTimes} from './components/feed-import-nytimes'
···
22
const wsurl = `${wsproto}://${wshost}/stream`
23
24
return (
25
+
<SkypodProvider
26
+
identityFallback={identityFallback}
27
+
connectionFallback={connectionFallback}
28
+
websocketUrl={wsurl}
29
+
>
30
+
<RealmConnectionManager />
31
+
<PeerList />
32
+
<FeedImportNYTimes />
33
+
<FeedImportTech />
34
+
<FeedImportPodcasts />
35
+
<Messenger />
36
+
<DebugNuke />
37
+
</SkypodProvider>
38
)
39
}
+150
src/client/skypod/action-dispatch/context.tsx
+150
src/client/skypod/action-dispatch/context.tsx
···
···
1
+
import {createContext} from 'preact'
2
+
import {useContext, useRef} from 'preact/hooks'
3
+
4
+
import {useDatabase} from '#client/root/context-database'
5
+
import {useRealmConnection} from '#realm/client/context-connection'
6
+
import {useRealmIdentity} from '#realm/client/context-identity'
7
+
import {LogicalClock} from '#realm/protocol/logical-clock'
8
+
9
+
import {Action, ActionMap, ActionOpts} from '#skypod/actions'
10
+
11
+
export type MiddlewareFn = (
12
+
this: undefined,
13
+
action: Action,
14
+
) => void | Action[] | Promise<void | Action[]>
15
+
16
+
export type MiddlewareRouterFn<K extends keyof ActionMap> = (
17
+
this: undefined,
18
+
action: ActionMap[K],
19
+
) => void | Action[] | Promise<void | Action[]>
20
+
21
+
export type MiddlewarePosition = 'push' | 'shift' | number
22
+
23
+
export type Middleware = MiddlewareFn | {[K in keyof ActionMap]?: MiddlewareRouterFn<K>}
24
+
25
+
export interface ActionDispatchContext {
26
+
action<N extends keyof ActionMap>(msg: N, dat: ActionMap[N]['dat'], opt?: ActionOpts): Action
27
+
dispatch(this: void, action: Action): Promise<void>
28
+
29
+
addMiddleware(this: void, handler: Middleware, position?: MiddlewarePosition): void
30
+
removeMiddleware(this: void, handler: Middleware): void
31
+
}
32
+
33
+
const ActionDispatchContext = createContext<ActionDispatchContext | null>(null)
34
+
35
+
export const ActionDispatchProvider: preact.FunctionComponent<{
36
+
children: preact.ComponentChildren
37
+
}> = (props) => {
38
+
const {db} = useDatabase()
39
+
const {identity} = useRealmIdentity()
40
+
const {realm} = useRealmConnection()
41
+
const middleware = useRef<Middleware[]>([])
42
+
43
+
const context = useRef<ActionDispatchContext>({
44
+
dispatch: async <K extends keyof ActionMap>(action: ActionMap[K]) => {
45
+
try {
46
+
const {identid: actor} = LogicalClock.extract(action.clk)
47
+
await db.transaction('rw', [db.clocks, db.actions], async (tx) => {
48
+
await db.actions.add({clock: action.clk, actor, action})
49
+
50
+
const extant = await tx.clocks.get(actor)
51
+
if (!extant || LogicalClock.compare(extant.clock, action.clk) < 0)
52
+
await tx.clocks.put({actor, clock: action.clk})
53
+
})
54
+
} catch (err: unknown) {
55
+
if (typeof err === 'object' && err && 'name' in err && err.name === 'ConstraintError') {
56
+
console.debug('duplicate action ignored', action)
57
+
} else {
58
+
console.error('failed to store action:', err)
59
+
}
60
+
61
+
return
62
+
}
63
+
64
+
const actions = [action] as Action[]
65
+
for (const action of actions) {
66
+
for (const mware of middleware.current) {
67
+
const mwarefn =
68
+
typeof mware === 'function'
69
+
? mware
70
+
: (mware[action.msg] as MiddlewareRouterFn<typeof action.msg> | undefined)
71
+
72
+
if (mwarefn) {
73
+
try {
74
+
const result = await Promise.resolve(mwarefn.call(undefined, action))
75
+
if (result !== void 0) {
76
+
actions.push(...result)
77
+
}
78
+
} catch (err: unknown) {
79
+
console.error(`middleware error for ${action.msg}:`, err)
80
+
continue // TODO: ???
81
+
}
82
+
}
83
+
}
84
+
}
85
+
86
+
// broadcast if not a local action
87
+
if (!action.opt?.local) {
88
+
realm.value?.broadcast([action], false)
89
+
}
90
+
},
91
+
92
+
action: <N extends keyof ActionMap>(
93
+
msg: N,
94
+
dat: ActionMap[N]['dat'],
95
+
opt?: ActionOpts,
96
+
): ActionMap[N] => {
97
+
const clk = identity.clock.now()
98
+
return {typ: 'act', clk, msg, dat, opt} as ActionMap[N]
99
+
},
100
+
101
+
addMiddleware: (handler, position) => {
102
+
if (middleware.current.indexOf(handler) !== -1) {
103
+
return
104
+
}
105
+
106
+
switch (position) {
107
+
case 'push':
108
+
case undefined:
109
+
middleware.current = [...middleware.current, handler]
110
+
break
111
+
112
+
case 'shift':
113
+
middleware.current = [handler, ...middleware.current]
114
+
break
115
+
116
+
default: {
117
+
if (middleware.current.length > position) {
118
+
const prefix = middleware.current.slice(0, position)
119
+
const suffix = middleware.current.slice(position)
120
+
121
+
middleware.current = [...prefix, handler, ...suffix]
122
+
}
123
+
}
124
+
}
125
+
},
126
+
127
+
removeMiddleware: (handler) => {
128
+
const index = middleware.current.indexOf(handler)
129
+
if (index >= 0) {
130
+
const prefix = middleware.current.slice(0, index)
131
+
const suffix = middleware.current.slice(index + 1)
132
+
133
+
middleware.current = [...prefix, ...suffix]
134
+
}
135
+
},
136
+
})
137
+
138
+
return (
139
+
<ActionDispatchContext.Provider value={context.current}>
140
+
{props.children}
141
+
</ActionDispatchContext.Provider>
142
+
)
143
+
}
144
+
145
+
export function useActionDispatch() {
146
+
const context = useContext(ActionDispatchContext)
147
+
if (!context) throw new Error('expected to be called inside an actions dispatch context!')
148
+
149
+
return context
150
+
}
+25
-241
src/client/skypod/context.tsx
+25
-241
src/client/skypod/context.tsx
···
1
-
import {createContext} from 'preact'
2
-
import {useContext, useEffect, useRef} from 'preact/hooks'
3
-
import {z} from 'zod/v4'
4
-
5
-
import {IdentID} from '#realm/protocol/index'
6
-
import {Action, ActionMap, ActionOpts, actionSchema} from '#skypod/actions'
7
-
import {feedSchema} from '#skypod/schema'
8
-
9
-
import {useDatabase} from '#client/root/context-database'
10
-
import {useRealmConnection} from '#realm/client/context-connection'
11
-
import {useRealmIdentity} from '#realm/client/context-identity'
12
-
13
-
import {LogicalClock} from '#realm/protocol/logical-clock'
14
-
import FeedFetchWorker from './feed-fetch.worker?worker'
15
-
import {createFeedMiddleware} from './middleware-feeds'
16
-
17
-
export type MiddlewareFn = (
18
-
this: undefined,
19
-
action: Action,
20
-
) => void | Action[] | Promise<void | Action[]>
21
-
22
-
export type MiddlewareRouterFn<K extends keyof ActionMap> = (
23
-
this: undefined,
24
-
action: ActionMap[K],
25
-
) => void | Action[] | Promise<void | Action[]>
26
-
27
-
export type MiddlewarePosition = 'push' | 'shift' | number
28
-
29
-
export type Middleware = MiddlewareFn | {[K in keyof ActionMap]?: MiddlewareRouterFn<K>}
30
-
31
-
export interface SkypodContext {
32
-
action<N extends keyof ActionMap>(msg: N, dat: ActionMap[N]['dat'], opt?: ActionOpts): Action
33
-
dispatch(this: void, action: Action): Promise<void>
34
-
35
-
addMiddleware(this: void, handler: Middleware, position?: MiddlewarePosition): void
36
-
removeMiddleware(this: void, handler: Middleware): void
37
-
}
38
-
39
-
const SkypodContext = createContext<SkypodContext | null>(null)
40
-
41
-
///
42
-
43
-
export const SkypodProvider: preact.FunctionComponent<{children: preact.ComponentChildren}> = (
44
-
props,
45
-
) => {
46
-
const {db} = useDatabase()
47
-
const {identity} = useRealmIdentity()
48
-
const {realm} = useRealmConnection()
49
-
50
-
const processor = useRef<Worker>(null)
51
-
52
-
// Initial middleware - feed management
53
-
const middleware = useRef<Middleware[]>([createFeedMiddleware(db, processor.current)])
54
-
55
-
const context = useRef<SkypodContext>({
56
-
dispatch: async <K extends keyof ActionMap>(action: ActionMap[K]) => {
57
-
try {
58
-
const {identid: actor} = LogicalClock.extract(action.clk)
59
-
await db.transaction('rw', [db.clocks, db.actions], async (tx) => {
60
-
await db.actions.add({clock: action.clk, actor, action})
61
-
62
-
const extant = await tx.clocks.get(actor)
63
-
if (!extant || LogicalClock.compare(extant.clock, action.clk) < 0)
64
-
await tx.clocks.put({actor, clock: action.clk})
65
-
})
66
-
} catch (err: unknown) {
67
-
if (typeof err === 'object' && err && 'name' in err && err.name === 'ConstraintError') {
68
-
console.debug('duplicate action ignored', action)
69
-
} else {
70
-
console.error('failed to store action:', err)
71
-
}
72
-
73
-
return
74
-
}
75
-
76
-
const actions = [action] as Action[]
77
-
for (const action of actions) {
78
-
for (const mware of middleware.current) {
79
-
const mwarefn =
80
-
typeof mware === 'function'
81
-
? mware
82
-
: (mware[action.msg] as MiddlewareRouterFn<typeof action.msg> | undefined)
83
-
84
-
if (mwarefn) {
85
-
try {
86
-
const result = await Promise.resolve(mwarefn.call(undefined, action))
87
-
if (result !== void 0) {
88
-
actions.push(...result)
89
-
}
90
-
} catch (err: unknown) {
91
-
console.error(`middleware error for ${action.msg}:`, err)
92
-
continue // TODO: ???
93
-
}
94
-
}
95
-
}
96
-
}
97
-
98
-
// broadcast if not a local action
99
-
if (!action.opt?.local) {
100
-
realm.value?.broadcast([action], false)
101
-
}
102
-
},
103
-
104
-
action: <N extends keyof ActionMap>(
105
-
msg: N,
106
-
dat: ActionMap[N]['dat'],
107
-
opt?: ActionOpts,
108
-
): ActionMap[N] => {
109
-
const clk = identity.clock.now()
110
-
return {typ: 'act', clk, msg, dat, opt} as ActionMap[N]
111
-
},
112
-
113
-
addMiddleware: (handler, position) => {
114
-
if (middleware.current.indexOf(handler) !== -1) {
115
-
return
116
-
}
117
-
118
-
switch (position) {
119
-
case 'push':
120
-
case undefined:
121
-
middleware.current = [...middleware.current, handler]
122
-
break
123
-
124
-
case 'shift':
125
-
middleware.current = [handler, ...middleware.current]
126
-
break
127
-
128
-
default: {
129
-
if (middleware.current.length > position) {
130
-
const prefix = middleware.current.slice(0, position)
131
-
const suffix = middleware.current.slice(position)
132
-
133
-
middleware.current = [...prefix, handler, ...suffix]
134
-
}
135
-
}
136
-
}
137
-
},
138
-
139
-
removeMiddleware: (handler) => {
140
-
const index = middleware.current.indexOf(handler)
141
-
if (index >= 0) {
142
-
const prefix = middleware.current.slice(0, index)
143
-
const suffix = middleware.current.slice(index + 1)
144
-
145
-
middleware.current = [...prefix, ...suffix]
146
-
}
147
-
},
148
-
})
149
150
-
// watch the connection
151
-
// while we're connected, watch peers for action messages
152
-
useEffect(() => {
153
-
const connection = realm.value
154
-
if (!connection) return
155
156
-
// we're connected, handle messages from peers
157
-
const handler = (event: CustomEvent<{identid: IdentID; data: unknown}>) => {
158
-
const go = async () => {
159
-
const json: unknown =
160
-
typeof event.detail.data === 'string' ? JSON.parse(event.detail.data) : event.detail.data
161
-
const data: unknown[] = Array.isArray(json) ? json : [json]
162
-
163
-
for (const datum of data) {
164
-
const parsed = actionSchema.safeParse(datum)
165
-
if (parsed.success) {
166
-
console.log('handling forwarded event:', parsed)
167
-
168
-
identity.clock.tick(parsed.data.clk)
169
-
await context.current.dispatch({...parsed.data, opt: {local: true}})
170
-
}
171
-
}
172
-
}
173
-
174
-
go().catch((err: unknown) => {
175
-
console.error(err)
176
-
return
177
-
})
178
-
}
179
-
180
-
connection.addEventListener('peerdata', handler as EventListener)
181
-
connection.addEventListener('wsdata', handler as EventListener)
182
-
return () => {
183
-
connection.removeEventListener('peerdata', handler as EventListener)
184
-
connection.removeEventListener('wsdata', handler as EventListener)
185
-
}
186
-
}, [context, identity, realm.value])
187
-
188
-
const patchSchema = z.union([
189
-
z.object({
190
-
msg: z.literal('patch'),
191
-
key: z.string(),
192
-
changes: feedSchema.partial(),
193
-
}),
194
-
z.object({
195
-
msg: z.literal('error'),
196
-
key: z.string(),
197
-
error: z.string(),
198
-
}),
199
-
])
200
-
201
-
// start feed processor worker
202
-
useEffect(() => {
203
-
const worker = new FeedFetchWorker()
204
-
205
-
worker.onmessage = async (event: MessageEvent) => {
206
-
const parsed = patchSchema.safeParse(event.data)
207
-
console.log('message from fetch worker', parsed)
208
-
209
-
switch (parsed.data?.msg) {
210
-
case 'patch': {
211
-
const action = context.current.action('feed:patch', {
212
-
url: parsed.data.key,
213
-
payload: parsed.data.changes,
214
-
})
215
-
console.log('sending action:', action)
216
-
await context.current.dispatch(action)
217
-
break
218
-
}
219
-
220
-
case 'error':
221
-
default:
222
-
console.error('unknown message from worker', parsed)
223
-
}
224
-
}
225
-
226
-
worker.onerror = (error) => {
227
-
console.error('Feed processor worker error:', error)
228
-
}
229
-
230
-
worker.postMessage({msg: 'start', identid: identity.identid})
231
-
232
-
processor.current = worker
233
-
return () => {
234
-
worker.terminate()
235
-
}
236
-
})
237
-
238
-
console.log('rendering the skypod context')
239
-
return <SkypodContext.Provider value={context.current}>{props.children}</SkypodContext.Provider>
240
}
241
242
-
export function useSkypod() {
243
-
const context = useContext(SkypodContext)
244
-
if (!context) throw new Error('expected to be called inside a database context!')
245
246
-
return context
247
}
···
1
+
import {DatabaseProvider} from '#client/root/context-database'
2
+
import {RealmConnectionFallbackProps, RealmConnectionProvider} from '#realm/client/context-connection'
3
+
import {RealmIdentityFallbackProps, RealmIdentityProvider} from '#realm/client/context-identity'
4
5
+
import {ActionDispatchProvider} from './action-dispatch/context'
6
+
import {EffectsConnection} from './effects-connection'
7
+
import {EffectsFeedProcessor} from './effects-feed-processor'
8
9
+
export interface SkypodProviderProps {
10
+
websocketUrl: string
11
+
identityFallback: (props: RealmIdentityFallbackProps) => preact.ComponentChild
12
+
connectionFallback: (props: RealmConnectionFallbackProps) => preact.ComponentChild
13
+
children: preact.ComponentChildren
14
}
15
16
+
export const SkypodProvider: preact.FunctionComponent<SkypodProviderProps> = (props) => {
17
+
return (
18
+
<DatabaseProvider>
19
+
<RealmIdentityProvider fallback={props.identityFallback}>
20
+
<RealmConnectionProvider fallback={props.connectionFallback} url={props.websocketUrl}>
21
+
<ActionDispatchProvider>
22
+
<EffectsConnection />
23
+
<EffectsFeedProcessor />
24
25
+
{props.children}
26
+
</ActionDispatchProvider>
27
+
</RealmConnectionProvider>
28
+
</RealmIdentityProvider>
29
+
</DatabaseProvider>
30
+
)
31
}
+52
src/client/skypod/effects-connection.tsx
+52
src/client/skypod/effects-connection.tsx
···
···
1
+
import {useEffect} from 'preact/hooks'
2
+
3
+
import {useRealmConnection} from '#realm/client/context-connection'
4
+
import {useRealmIdentity} from '#realm/client/context-identity'
5
+
import {IdentID} from '#realm/protocol/index'
6
+
7
+
import {actionSchema} from '#skypod/actions'
8
+
import {useActionDispatch} from './action-dispatch/context'
9
+
10
+
export const EffectsConnection: preact.FunctionComponent = () => {
11
+
const {identity} = useRealmIdentity()
12
+
const {realm} = useRealmConnection()
13
+
const dispatcher = useActionDispatch()
14
+
15
+
// watch the connection
16
+
// while we're connected, watch peers for action messages
17
+
18
+
useEffect(() => {
19
+
const connection = realm.value
20
+
if (!connection) return
21
+
22
+
const handler = (event: CustomEvent<{identid: IdentID; data: unknown}>) => {
23
+
const go = async () => {
24
+
const json: unknown = typeof event.detail.data === 'string' ? JSON.parse(event.detail.data) : event.detail.data
25
+
const data: unknown[] = Array.isArray(json) ? json : [json]
26
+
27
+
for (const datum of data) {
28
+
const parsed = actionSchema.safeParse(datum)
29
+
if (parsed.success) {
30
+
console.log('handling forwarded event:', parsed)
31
+
32
+
identity.clock.tick(parsed.data.clk)
33
+
await dispatcher.dispatch({...parsed.data, opt: {local: true}})
34
+
}
35
+
}
36
+
}
37
+
38
+
go().catch((exc: unknown) => {
39
+
console.error(exc)
40
+
})
41
+
}
42
+
43
+
connection.addEventListener('peerdata', handler as EventListener)
44
+
connection.addEventListener('wsdata', handler as EventListener)
45
+
return () => {
46
+
connection.removeEventListener('peerdata', handler as EventListener)
47
+
connection.removeEventListener('wsdata', handler as EventListener)
48
+
}
49
+
}, [dispatcher, identity, realm.value])
50
+
51
+
return <></>
52
+
}
+70
src/client/skypod/effects-feed-processor.tsx
+70
src/client/skypod/effects-feed-processor.tsx
···
···
1
+
import {useEffect} from 'preact/hooks'
2
+
import {z} from 'zod/v4'
3
+
4
+
import {useDatabase} from '#client/root/context-database'
5
+
import {Action} from '#skypod/actions'
6
+
7
+
import {useActionDispatch} from './action-dispatch/context'
8
+
9
+
import {useRealmIdentity} from '#realm/client/context-identity.js'
10
+
import {ReqEvent, respSchema} from './feed-processor/api'
11
+
import {createFeedMiddleware} from './feed-processor/middleware'
12
+
import FeedFetchWorker from './feed-processor/worker?worker'
13
+
14
+
export const EffectsFeedProcessor: preact.FunctionComponent = () => {
15
+
const {db} = useDatabase()
16
+
const {identity} = useRealmIdentity()
17
+
const dispatcher = useActionDispatch()
18
+
19
+
useEffect(() => {
20
+
const worker = new FeedFetchWorker()
21
+
22
+
worker.onmessage = async (event: MessageEvent) => {
23
+
const parsed = respSchema.safeParse(event.data)
24
+
if (!parsed.success) {
25
+
console.warn('unknown message from feed worker?', z.treeifyError(parsed.error))
26
+
return
27
+
}
28
+
29
+
let action: Action
30
+
switch (parsed.data.msg) {
31
+
case 'feed:patch':
32
+
action = dispatcher.action('feed:patch', {
33
+
url: parsed.data.dat.feedurl,
34
+
payload: parsed.data.dat.changes,
35
+
})
36
+
break
37
+
38
+
default:
39
+
console.log('not handled yet:', parsed.data)
40
+
return
41
+
}
42
+
43
+
console.log('sending action:', action)
44
+
await dispatcher.dispatch(action)
45
+
}
46
+
47
+
worker.onerror = (error: unknown) => {
48
+
console.error('Feed processor worker error:', error)
49
+
}
50
+
51
+
// attach the middleware
52
+
const middleware = createFeedMiddleware(db, worker)
53
+
dispatcher.addMiddleware(middleware)
54
+
55
+
// kick it off only after we've attached middleware to handle the results
56
+
worker.postMessage({
57
+
typ: 'evt',
58
+
msg: 'init',
59
+
dat: {identid: identity.identid},
60
+
} satisfies ReqEvent)
61
+
62
+
// shut it down when we're done
63
+
return () => {
64
+
dispatcher.removeMiddleware(middleware)
65
+
worker.terminate()
66
+
}
67
+
}, [identity, db, dispatcher])
68
+
69
+
return <></>
70
+
}
+46
-46
src/client/skypod/feed-fetch.worker.ts
src/client/skypod/feed-processor/worker.ts
+46
-46
src/client/skypod/feed-fetch.worker.ts
src/client/skypod/feed-processor/worker.ts
···
1
import {IndexableType} from 'dexie'
2
-
import {z} from 'zod/v4'
3
4
import {Database} from '#client/root/service-database'
5
import {normalizeProtocolError} from '#common/errors'
6
-
import {IdentBrand, IdentID} from '#realm/protocol/index'
7
import {LCTimestamp, LogicalClock} from '#realm/protocol/logical-clock'
8
9
-
const msgStartSchema = z.object({
10
-
msg: z.literal('start'),
11
-
identid: IdentBrand.schema,
12
-
})
13
14
-
const msgPollSchema = z.object({
15
-
msg: z.literal('poll'),
16
-
})
17
18
-
const msgStopSchema = z.object({
19
-
msg: z.literal('stop'),
20
-
})
21
22
-
const msgSchema = z.discriminatedUnion('msg', [msgStartSchema, msgPollSchema, msgStopSchema])
23
24
class FeedFetch {
25
#db: Database
26
#owner: IdentID
27
#clock: LogicalClock
28
-
#timeout: ReturnType<typeof setTimeout>
29
30
constructor(identid: IdentID) {
31
this.#db = new Database()
32
this.#clock = new LogicalClock(identid)
33
this.#owner = identid
34
35
-
this.#timeout = setTimeout(this.#poll, 10000)
36
}
37
38
stop() {
39
-
clearTimeout(this.#timeout)
40
}
41
42
-
poll() {
43
-
clearTimeout(this.#timeout)
44
-
this.#poll()
45
}
46
47
#poll = () => {
48
-
const pendingFeeds = this.#db.feeds.where('lastRefresh.status').equals('pending')
49
-
this.#db
50
-
.withLock('feeds', pendingFeeds, this.#clock, this.#owner, this.#pollLocked)
51
.catch((ex: unknown) => {
52
console.error('problem locking pending feeds', ex)
53
})
···
56
})
57
}
58
59
-
#pollLocked = async (urls: IndexableType[], lock: LCTimestamp) => {
60
console.log('checking feeds...', urls, lock)
61
62
try {
···
107
} catch (ex: unknown) {
108
console.error('problem fetching pending feeds:', ex)
109
}
110
-
}
111
-
}
112
-
113
-
let fetcher: FeedFetch
114
-
115
-
onmessage = (event: MessageEvent) => {
116
-
const parsed = msgSchema.safeParse(event.data)
117
-
switch (parsed.data?.msg) {
118
-
case 'start':
119
-
fetcher = new FeedFetch(parsed.data.identid)
120
-
break
121
-
122
-
case 'poll':
123
-
fetcher.poll()
124
-
break
125
-
126
-
case 'stop':
127
-
fetcher.stop()
128
-
break
129
-
130
-
default:
131
-
console.warn('unknown message, bailing', event.data, parsed.error)
132
-
return
133
}
134
}
135
···
1
import {IndexableType} from 'dexie'
2
3
import {Database} from '#client/root/service-database'
4
import {normalizeProtocolError} from '#common/errors'
5
+
import {IdentID} from '#realm/protocol/index'
6
import {LCTimestamp, LogicalClock} from '#realm/protocol/logical-clock'
7
8
+
import {reqSchema} from './api'
9
10
+
let instance: FeedFetch | undefined
11
+
onmessage = (event: MessageEvent) => {
12
+
const parsed = reqSchema.safeParse(event.data)
13
+
if (!parsed.success) return // unexpected (preact sends page events)
14
15
+
switch (parsed.data.msg) {
16
+
case 'init':
17
+
instance = new FeedFetch(parsed.data.dat.identid)
18
+
break
19
20
+
case 'work':
21
+
instance?.processUrls(parsed.data.dat.urls).catch((exc: unknown) => {
22
+
console.error('error processing urls:', exc)
23
+
})
24
+
break
25
+
}
26
+
}
27
28
class FeedFetch {
29
#db: Database
30
#owner: IdentID
31
#clock: LogicalClock
32
+
#timeout?: ReturnType<typeof setTimeout>
33
+
34
+
// worker, so we have it get it's own db and clock
35
36
constructor(identid: IdentID) {
37
this.#db = new Database()
38
this.#clock = new LogicalClock(identid)
39
this.#owner = identid
40
41
+
this.#poll()
42
}
43
44
stop() {
45
+
if (this.#timeout) clearTimeout(this.#timeout)
46
}
47
48
+
async processPending() {
49
+
const pendingFeeds = this.#db.feeds.where('lastRefresh.status').equals('pending')
50
+
return await this.#db.withLock(
51
+
'feeds',
52
+
pendingFeeds,
53
+
this.#clock,
54
+
this.#owner,
55
+
this.#processLocked,
56
+
)
57
+
}
58
+
59
+
async processUrls(urls: string[]) {
60
+
const requestedFeeds = this.#db.feeds.where('url').anyOf(urls)
61
+
return await this.#db.withLock(
62
+
'feeds',
63
+
requestedFeeds,
64
+
this.#clock,
65
+
this.#owner,
66
+
this.#processLocked,
67
+
)
68
}
69
70
#poll = () => {
71
+
if (this.#timeout) clearTimeout(this.#timeout)
72
+
73
+
this.processPending()
74
.catch((ex: unknown) => {
75
console.error('problem locking pending feeds', ex)
76
})
···
79
})
80
}
81
82
+
#processLocked = async (urls: IndexableType[], lock: LCTimestamp) => {
83
console.log('checking feeds...', urls, lock)
84
85
try {
···
130
} catch (ex: unknown) {
131
console.error('problem fetching pending feeds:', ex)
132
}
133
}
134
}
135
+42
src/client/skypod/feed-processor/api.ts
+42
src/client/skypod/feed-processor/api.ts
···
···
1
+
import {IdentBrand} from '#realm/protocol'
2
+
import {makeEventSchema} from '#realm/protocol/schema.js'
3
+
import {z} from 'zod/v4'
4
+
5
+
import {feedEntrySchema, feedSchema} from '#skypod/schema'
6
+
7
+
export const initEvent = makeEventSchema(
8
+
'init',
9
+
z.object({
10
+
identid: IdentBrand.schema,
11
+
}),
12
+
)
13
+
14
+
export const workEvent = makeEventSchema(
15
+
'work',
16
+
z.object({
17
+
urls: z.array(z.url()),
18
+
}),
19
+
)
20
+
21
+
export const reqSchema = z.union([initEvent, workEvent])
22
+
export type ReqEvent = z.infer<typeof reqSchema>
23
+
24
+
export const patchFeedEvent = makeEventSchema(
25
+
'feed:patch',
26
+
z.object({
27
+
feedurl: z.string(),
28
+
changes: feedSchema.partial(),
29
+
}),
30
+
)
31
+
32
+
export const patchFeedEntryEvent = makeEventSchema(
33
+
'feedentry:patch',
34
+
z.object({
35
+
feedurl: z.string(),
36
+
entryguid: z.string(),
37
+
changes: feedEntrySchema.partial(),
38
+
}),
39
+
)
40
+
41
+
export const respSchema = z.union([patchFeedEvent, patchFeedEntryEvent])
42
+
export type RespEvent = z.infer<typeof respSchema>
+12
-10
src/client/skypod/middleware-feeds.ts
src/client/skypod/feed-processor/middleware.ts
+12
-10
src/client/skypod/middleware-feeds.ts
src/client/skypod/feed-processor/middleware.ts
···
1
import {Database} from '#client/root/service-database'
2
-
import {Middleware} from './context'
3
4
-
/**
5
-
* Feed management middleware
6
-
*
7
-
* Handles feed lifecycle actions:
8
-
* - feed:add - Creates feed record and triggers fetch
9
-
* - feed:remove - Deletes feed from database
10
-
* - feed:patch - Updates feed metadata
11
-
*/
12
export function createFeedMiddleware(db: Database, worker: Worker | null): Middleware {
13
return {
14
'feed:add': async (action) => {
···
26
},
27
})
28
29
-
worker?.postMessage({msg: 'poll'})
30
},
31
32
'feed:remove': async (action) => {
···
35
36
'feed:patch': async (action) => {
37
await db.feeds.update(action.dat.url, action.dat.payload)
38
},
39
}
40
}
···
1
import {Database} from '#client/root/service-database'
2
+
import {Middleware} from '../action-dispatch/context'
3
4
+
import {ReqEvent} from './api'
5
+
6
export function createFeedMiddleware(db: Database, worker: Worker | null): Middleware {
7
return {
8
'feed:add': async (action) => {
···
20
},
21
})
22
23
+
worker?.postMessage({
24
+
typ: 'evt',
25
+
msg: 'work',
26
+
dat: {urls: [action.dat.url]},
27
+
} satisfies ReqEvent)
28
},
29
30
'feed:remove': async (action) => {
···
33
34
'feed:patch': async (action) => {
35
await db.feeds.update(action.dat.url, action.dat.payload)
36
+
},
37
+
38
+
'feedentry:patch': (action) => {
39
+
console.log('feedentry patch', action)
40
},
41
}
42
}
+1
-1
src/server/routes-api/middleware-cors.ts
+1
-1
src/server/routes-api/middleware-cors.ts
+13
-11
src/server/routes-api/middleware-reader.ts
+13
-11
src/server/routes-api/middleware-reader.ts
···
1
import {Readability} from '@mozilla/readability'
2
-
import {RequestHandler} from "express"
3
import {parseHTML} from 'linkedom'
4
5
export const readabilityProxy: RequestHandler = async (req, res) => {
···
36
}
37
38
res.setHeader('Content-Type', 'application/json; charset=utf-8')
39
-
res.send(JSON.stringify({
40
-
title: article.title,
41
-
byline: article.byline,
42
-
content: article.content,
43
-
textContent: article.textContent,
44
-
excerpt: article.excerpt,
45
-
siteName: article.siteName,
46
-
length: article.length,
47
-
url,
48
-
}))
49
} catch (error) {
50
console.error('Reader mode error:', error)
51
res.status(500).json({
···
1
import {Readability} from '@mozilla/readability'
2
+
import {RequestHandler} from 'express'
3
import {parseHTML} from 'linkedom'
4
5
export const readabilityProxy: RequestHandler = async (req, res) => {
···
36
}
37
38
res.setHeader('Content-Type', 'application/json; charset=utf-8')
39
+
res.send(
40
+
JSON.stringify({
41
+
title: article.title,
42
+
byline: article.byline,
43
+
content: article.content,
44
+
textContent: article.textContent,
45
+
excerpt: article.excerpt,
46
+
siteName: article.siteName,
47
+
length: article.length,
48
+
url,
49
+
}),
50
+
)
51
} catch (error) {
52
console.error('Reader mode error:', error)
53
res.status(500).json({
+2
-2
src/server/routes-api/middleware.ts
+2
-2
src/server/routes-api/middleware.ts