1// lib/flux.ts
2//
3// A complete, single file reference wiring for an app with:
4// - a public 'global' scoped channel (site wide announcements)
5// - a private 'user' scoped list channel with full CRUD + offline queue
6// - a private 'user' scoped single record settings channel
7// - a singleton engine instance safe under Next.js Fast Refresh
8//
9// Copy this file as a starting point and swap the store adapters,
10// channel names, and endpoints for your own.
11
12import {
13 createFlux,
14 addQueueEntrySafe,
15 getAllQueueEntries,
16 removeQueueEntry,
17} from '@tsworldtech/flux'
18import { createSupabaseAdapter } from '@tsworldtech/flux-supabase'
19import { createClient } from './supabase/client'
20import { useAnnouncementsStore } from '../store/useAnnouncementsStore'
21import { useItemsStore } from '../store/useItemsStore'
22import { useSettingsStore, settingsToRow } from '../store/useSettingsStore'
23
24// 1. Singleton engine
25// Guards against Fast Refresh / multiple import re instantiation by
26// stashing the instance on window.
27const STORAGE_PREFIX = 'MyApp'
28const GLOBAL_KEY = '__flux_engine_singleton__'
29
30function getOrCreateFlux() {
31 if (typeof window !== 'undefined' && (window as any)[GLOBAL_KEY]) {
32 return (window as any)[GLOBAL_KEY]
33 }
34
35 const supabase = createClient()
36
37 const instance = createFlux({
38 adapter: createSupabaseAdapter(supabase),
39 license: process.env.NEXT_PUBLIC_FLUX_LICENSE_KEY,
40 storagePrefix: STORAGE_PREFIX,
41 ttl: {
42 short: 1000 * 60 * 5,
43 medium: 1000 * 60 * 60,
44 long: 1000 * 60 * 60 * 24,
45 },
46 onStorageFallback: (layer, reason) => {
47 console.warn('[flux] Storage degraded to', layer, '—', reason)
48 },
49 onConflictUnified: (conflicts) => {
50 // Drive your own conflict resolution UI from this single callback.
51 // See core_concepts/conflict_resolution for the full contract.
52 },
53 })
54
55 if (typeof window !== 'undefined') (window as any)[GLOBAL_KEY] = instance
56 return instance
57}
58
59export const flux = getOrCreateFlux()
60
61// 1b. Realtime cluster boot
62// Starts leader election across open tabs and activates the realtime
63// connection. Without it the engine is registered and IDB is populated
64// by bootstrap, but the realtime manager stays inert: no leader is
65// elected and no channel is opened, so live push AND the catch-up sync
66// after a reconnect never fire. Guarded by a window flag so any number
67// of callers trigger it only once per tab. It is called from the
68// bootstraps in Section 4 — after a confirmed bootstrap, never before.
69
70const CLUSTER_BOOT_KEY = '__flux_cluster_booted__'
71
72export function ensureClusterBootOnce(): void {
73 if (typeof window === 'undefined') return
74 if (window[CLUSTER_BOOT_KEY]) return
75 window[CLUSTER_BOOT_KEY] = true
76 flux.realtime.ensureClusterBoot()
77}
78
79// 2. Store adapters
80// Every registered channel needs a { getState, setState } shape. Native
81// Zustand hooks already have this — see state_managers/zustand for Redux,
82// Jotai, and Stunk equivalents.
83
84export const announcementsStoreAdapter = {
85 setState: (data: any) => useAnnouncementsStore.getState().setState(data),
86 getState: () => useAnnouncementsStore.getState().announcements,
87}
88
89export const itemsStoreAdapter = {
90 setState: (data: any) => useItemsStore.getState().setState(data),
91 getState: () => useItemsStore.getState().items,
92}
93
94export const settingsStoreAdapter = {
95 setState: (data: any) => useSettingsStore.getState().setState(data),
96 getState: () => {
97 const s = useSettingsStore.getState().settings
98 return s ? [s] : []
99 },
100}
101
102// 3. Registrations
103// Global scoped registrations are safe behind a PERMANENT guard: run once
104// per page load, and never touched again, because flux.clearUserSession()
105// (Section 7) never unregisters 'global' scoped channels — see
106// core_concepts/multi_tenant_scoping.
107//
108// 'user' scoped registrations are different. clearUserSession() DOES
109// unregister every 'user' scoped channel as part of its purge — that's
110// correct, it's how Flux knows what to sweep from IndexedDB on logout.
111// But it means a 'user' scoped flux.register() call guarded by the SAME
112// permanent flag as the block below only ever runs once, on the first
113// page load: after the first logout the registry is empty for the rest
114// of the tab's life. Every bootstrap() call after that resolves without
115// error but silently finds no channel to dispatch into, and the UI ends
116// up marking dataReady: true against nothing — the most common cause of
117// "logged in successfully, but the dashboard is empty".
118//
119// registerUserChannels() below solves this with its own idempotent,
120// SESSION scoped guard (a plain module variable, not a permanent window
121// flag) that markUserChannelsUnregistered() flips back off. Call
122// registerUserChannels() right before every bootstrapUserData() call
123// (Section 4), and markUserChannelsUnregistered() right alongside every
124// clearUserSession() call (Section 7).
125
126const REGISTERED_KEY = '__flux_registered__'
127let userChannelsRegistered = false
128
129declare global {
130 interface Window {
131 __flux_engine_singleton__?: any
132 __flux_registered__?: boolean
133 __flux_cluster_booted__?: boolean
134 }
135}
136
137if (typeof window !== 'undefined' && !window[REGISTERED_KEY]) {
138 window[REGISTERED_KEY] = true
139
140 // Public, unscoped — survives logout indefinitely, visible to anyone.
141 // Safe behind a permanent guard because this channel is never
142 // unregistered by clearUserSession().
143 flux.register({
144 channel: 'announcements',
145 table: 'announcements',
146 idbKey: 'announcements:list',
147 ttl: 'long',
148 ingestionType: 'COLLECTION_ALL',
149 diffBeforeUpdate: true,
150 event: 'UPDATE',
151 store: announcementsStoreAdapter,
152 // no scope field — defaults to 'global'
153 })
154}
155
156// Call this once per login, right before bootstrapping user scoped data —
157// never behind a permanent window flag. Idempotent within a session: a
158// second call before the next markUserChannelsUnregistered() is a no op,
159// so it's cheap and safe to call on every login attempt.
160export function registerUserChannels(): void {
161 if (userChannelsRegistered) return
162 userChannelsRegistered = true
163
164 // Private list, full CRUD, offline safe.
165 flux.register({
166 channel: 'items',
167 table: 'items',
168 idbKey: 'items:list',
169 ttl: 'long',
170 ingestionType: 'COLLECTION_ALL',
171 diffBeforeUpdate: true,
172 event: '*',
173 scope: 'user',
174 store: itemsStoreAdapter,
175
176 queueConfig: {
177 storeName: 'items_mutation_queue',
178 replayEndpoint: '/api/items',
179 method: 'POST',
180 replayStrategy: 'fire_and_forget',
181 dedupeField: 'id',
182 dedupeStrategy: 'replace',
183
184 onReplaySuccess: async (responseBody, entry) => {
185 const method = (entry?.method ?? 'POST').toUpperCase()
186
187 if (method === 'DELETE') {
188 await flux.dispatchLocalUpdate('items', { op: 'DELETE', id: entry.payload?.id })
189 return
190 }
191
192 const confirmed = responseBody?.item
193 if (confirmed) await flux.dispatchLocalUpdate('items', confirmed)
194 },
195 },
196 })
197
198 // Private single record settings, no queue — settings writes overwrite
199 // an existing server row, so they use dispatchLocalUpdate after a
200 // synchronous confirmed save instead of the offline queue. See
201 // state_managers/zustand Step 7 for the full save flow this pairs with.
202 flux.register({
203 channel: 'settings',
204 table: 'user_settings',
205 idbKey: 'settings:mine',
206 ttl: 'medium',
207 ingestionType: 'SNAPSHOT',
208 diffBeforeUpdate: true,
209 event: 'UPDATE',
210 scope: 'user',
211 store: settingsStoreAdapter,
212 })
213}
214
215// Flips registerUserChannels()'s guard back off so the NEXT login
216// re registers everything clearUserSession() is about to tear down.
217// Call this alongside every flux.clearUserSession() — see Section 7.
218export function markUserChannelsUnregistered(): void {
219 userChannelsRegistered = false
220}
221
222// 4. Bootstrap
223// One call per scope. Fire the global bootstrap on app mount; fire the
224// user bootstrap once auth resolves. Each one starts the realtime cluster
225// once its own bootstrap has succeeded. See api/bootstrap for the full
226// leader election and revalidation behavior this triggers.
227
228export async function bootstrapGlobalData(): Promise<void> {
229 await flux.bootstrap({
230 endpoint: '/api/bootstrap/public',
231 scope: 'global',
232 map: {
233 announcements: announcementsStoreAdapter,
234 },
235 })
236
237 // Public data: safe to start realtime as soon as the seed lands.
238 ensureClusterBootOnce()
239}
240
241let activeUserBootstrap: Promise<void> | null = null
242let activeUserBootstrapId: string | null = null
243
244// Bumped on every logout (Section 7). A bootstrap that started before the
245// logout compares the value it captured at start against this when it
246// finishes, and discards its result instead of starting realtime for a
247// session that no longer exists.
248let purgeGeneration = 0
249
250export async function bootstrapUserData(userId: string): Promise<void> {
251 if (activeUserBootstrap && activeUserBootstrapId === userId) {
252 return activeUserBootstrap
253 }
254
255 activeUserBootstrapId = userId
256 activeUserBootstrap = (async () => {
257 const generationAtStart = purgeGeneration
258
259 try {
260 // Re register this session's user scoped channels before every
261 // bootstrap call — not just once, the first time this module loads.
262 // Idempotent within a session (Section 3), so calling it here on
263 // every login attempt is cheap and safe, and it's what actually
264 // closes the gap left by logout()'s markUserChannelsUnregistered()
265 // call in Section 7.
266 registerUserChannels()
267
268 await flux.bootstrap({
269 endpoint: '/api/bootstrap/user',
270 scope: 'user',
271 userId,
272 map: {
273 items: itemsStoreAdapter,
274 settings: settingsStoreAdapter,
275 },
276 })
277
278 if (purgeGeneration !== generationAtStart) {
279 console.log('[flux] Discarding stale bootstrap — a logout happened while it was in flight')
280 return
281 }
282
283 // bootstrap() only hydrates IDB over a plain HTTP call. Start
284 // realtime only after a confirmed bootstrap for THIS user — user
285 // scoped data must not be subscribed to before we know who the
286 // user is.
287 ensureClusterBootOnce()
288 } finally {
289 activeUserBootstrap = null
290 activeUserBootstrapId = null
291 }
292 })()
293
294 return activeUserBootstrap
295}
296
297// 5. Direct or queued writes
298// The pattern every mutation on the 'items' channel follows: try direct,
299// fall back to the queue, confirm once either path succeeds.
300
301const ITEMS_QUEUE_STORE = 'items_mutation_queue'
302
303async function postItemDirect(payload: { title: string }): Promise<{ ok: boolean; item?: any; error?: string }> {
304 try {
305 const res = await fetch('/api/items', {
306 method: 'POST',
307 headers: { 'Content-Type': 'application/json' },
308 body: JSON.stringify(payload),
309 })
310 const body = await res.json().catch(() => null)
311 if (!res.ok || !body?.ok) return { ok: false, error: body?.error || `Request failed (${res.status})` }
312 return { ok: true, item: body.item }
313 } catch (e: any) {
314 return { ok: false, error: e?.message || 'Network error' }
315 }
316}
317
318async function deleteItemDirect(id: string): Promise<{ ok: boolean; error?: string }> {
319 try {
320 const res = await fetch('/api/items', {
321 method: 'DELETE',
322 headers: { 'Content-Type': 'application/json' },
323 body: JSON.stringify({ id }),
324 })
325 if (!res.ok) return { ok: false, error: `Request failed (${res.status})` }
326 return { ok: true }
327 } catch (e: any) {
328 return { ok: false, error: e?.message || 'Network error' }
329 }
330}
331
332async function queueItemCreate(tempId: string, title: string): Promise<void> {
333 await addQueueEntrySafe(
334 {
335 channel: 'items',
336 storeName: ITEMS_QUEUE_STORE,
337 endpoint: '/api/items',
338 method: 'POST',
339 payload: { title, tempId },
340 timestamp: Date.now(),
341 } as any,
342 STORAGE_PREFIX,
343 undefined, undefined, undefined,
344 'user',
345 )
346}
347
348async function queueItemDelete(id: string): Promise<void> {
349 await addQueueEntrySafe(
350 {
351 channel: 'items',
352 storeName: ITEMS_QUEUE_STORE,
353 endpoint: '/api/items',
354 method: 'DELETE',
355 payload: { id },
356 timestamp: Date.now(),
357 dedupeKey: `item:${id}`,
358 dedupeStrategy: 'replace',
359 } as any,
360 STORAGE_PREFIX,
361 undefined, undefined, undefined,
362 'user',
363 )
364}
365
366async function purgePendingCreatesForItem(id: string): Promise<void> {
367 const entries = await getAllQueueEntries(STORAGE_PREFIX, ITEMS_QUEUE_STORE)
368 for (const entry of entries) {
369 const isCreate = (entry.method ?? 'POST') !== 'DELETE'
370 const entryId = String(entry.payload?.id ?? entry.payload?.tempId ?? '')
371 if (isCreate && entryId === id) {
372 await removeQueueEntry(entry.id, STORAGE_PREFIX, ITEMS_QUEUE_STORE)
373 }
374 }
375}
376
377export async function createItemAndSync(
378 title: string
379): Promise<{ error: Error | null; queued: boolean }> {
380 const tempId = `temp_item_${Date.now()}`
381
382 // Optimistic insert
383 await flux.dispatchLocalUpdate('items', { id: tempId, title, status: 'pending' })
384
385 const online = typeof navigator !== 'undefined' && navigator.onLine
386
387 if (online) {
388 const direct = await postItemDirect({ title })
389 if (!direct.ok) {
390 await queueItemCreate(tempId, title)
391 return { error: null, queued: true }
392 }
393 await flux.dispatchLocalUpdate('items', { op: 'DELETE', id: tempId })
394 await flux.dispatchLocalUpdate('items', direct.item)
395 return { error: null, queued: false }
396 }
397
398 await queueItemCreate(tempId, title)
399 return { error: null, queued: true }
400}
401
402export async function deleteItemAndSync(
403 id: string
404): Promise<{ error: Error | null; queued: boolean }> {
405 await flux.dispatchLocalUpdate('items', { op: 'DELETE', id })
406
407 const isTempId = id.startsWith('temp_item_')
408 if (isTempId) {
409 await purgePendingCreatesForItem(id)
410 return { error: null, queued: false }
411 }
412
413 await purgePendingCreatesForItem(id)
414
415 const online = typeof navigator !== 'undefined' && navigator.onLine
416
417 if (online) {
418 const direct = await deleteItemDirect(id)
419 if (!direct.ok) {
420 await queueItemDelete(id)
421 return { error: null, queued: true }
422 }
423 return { error: null, queued: false }
424 }
425
426 await queueItemDelete(id)
427 return { error: null, queued: true }
428}
429
430// 6. Settings save (SNAPSHOT channel, no queue)
431
432export async function saveSettingsAndSync(): Promise<void> {
433 const before = useSettingsStore.getState()
434 if (!before.isDirty()) return
435
436 let confirmedRow: any
437 try {
438 const res = await fetch('/api/settings', {
439 method: 'PATCH',
440 headers: { 'Content-Type': 'application/json' },
441 body: JSON.stringify(settingsToRow(before.settings)),
442 })
443 confirmedRow = await res.json()
444 } catch (err) {
445 console.warn('[settings] Save failed:', err)
446 return
447 }
448
449 await flux.dispatchLocalUpdate('settings', confirmedRow)
450}
451
452// 7. Session teardown
453// Call this from your logout handler and/or an auth state change listener.
454// It purges every 'user' scoped store, queue entry, sync anchor, and
455// tracked job — 'global' scoped data (announcements) is untouched.
456//
457// purgeGeneration is bumped first so any user bootstrap still in flight
458// discards its own result when it lands. markUserChannelsUnregistered()
459// runs next so registerUserChannels()'s guard is back off before the
460// next login attempt — otherwise the very next bootstrapUserData() call
461// sees userChannelsRegistered still true, skips re registering, and
462// dispatches into a registry that clearUserSession() just emptied.
463
464export async function logout(): Promise<void> {
465 purgeGeneration++
466 markUserChannelsUnregistered()
467 await flux.clearUserSession()
468}