625 lines
20 KiB
TypeScript
625 lines
20 KiB
TypeScript
/**
|
|
* Background sync service that scans all notes with service='zapwall.fr' tag
|
|
* and caches them in IndexedDB
|
|
* Runs in background (non-blocking) and updates cache when new notes are published
|
|
*/
|
|
|
|
import type { Event } from 'nostr-tools'
|
|
import type { SimplePoolWithSub } from '@/types/nostr-tools-extended'
|
|
import { createSubscription } from '@/types/nostr-tools-extended'
|
|
import { nostrService } from './nostr'
|
|
import { PLATFORM_SERVICE, MIN_EVENT_DATE } from './platformConfig'
|
|
import { extractTagsFromEvent } from './nostrTagSystem'
|
|
import { parsePresentationEvent } from './articlePublisherHelpersPresentation'
|
|
import { parseArticleFromEvent, parseSeriesFromEvent, parseReviewFromEvent, parsePurchaseFromEvent, parseReviewTipFromEvent, parseSponsoringFromEvent } from './nostrEventParsing'
|
|
|
|
const TARGET_EVENT_ID = '527d83e0af20bf23c3e104974090ccc21536ece72c24eb784b3642890f63b763'
|
|
|
|
class PlatformSyncService {
|
|
private syncInProgress = false
|
|
private syncSubscription: { unsub: () => void } | null = null
|
|
private lastSyncTime: number = 0
|
|
private readonly SYNC_INTERVAL_MS = 60000 // Sync every minute
|
|
private readonly SYNC_TIMEOUT_MS = 60000 // 60 seconds timeout per relay (increased from 30s)
|
|
|
|
/**
|
|
* Start background sync
|
|
* Scans all notes with service='zapwall.fr' and caches them
|
|
* Does not require pubkey - syncs all public notes with service='zapwall.fr'
|
|
*/
|
|
async startSync(): Promise<void> {
|
|
if (this.syncInProgress) {
|
|
return
|
|
}
|
|
|
|
const pool = nostrService.getPool()
|
|
if (!pool) {
|
|
console.warn('Pool not initialized, cannot start platform sync')
|
|
return
|
|
}
|
|
|
|
this.syncInProgress = true
|
|
|
|
// Update progress manager to show sync indicator
|
|
const { syncProgressManager } = await import('./syncProgressManager')
|
|
const { relaySessionManager } = await import('./relaySessionManager')
|
|
const activeRelays = await relaySessionManager.getActiveRelays()
|
|
const initialRelay = activeRelays[0] ?? 'Connecting...'
|
|
const totalRelays = activeRelays.length ?? 1
|
|
syncProgressManager.setProgress({ currentStep: 0, totalSteps: totalRelays, completed: false, currentRelay: initialRelay })
|
|
|
|
try {
|
|
await this.performSync(pool as unknown as SimplePoolWithSub)
|
|
|
|
// Mark as completed after all relays are processed
|
|
const finalRelay = activeRelays[activeRelays.length - 1] ?? initialRelay
|
|
syncProgressManager.setProgress({ currentStep: totalRelays, totalSteps: totalRelays, completed: true, currentRelay: finalRelay })
|
|
} catch (error) {
|
|
console.error('Error in platform sync:', error)
|
|
syncProgressManager.setProgress(null)
|
|
} finally {
|
|
this.syncInProgress = false
|
|
// Clear progress after a short delay
|
|
setTimeout(() => {
|
|
syncProgressManager.setProgress(null)
|
|
}, 500)
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Perform a sync operation
|
|
* Scans all notes with service='zapwall.fr' tag from ALL active relays
|
|
* Starts from January 5, 2026 00:00:00 UTC
|
|
*/
|
|
private async performSync(pool: SimplePoolWithSub): Promise<void> {
|
|
// Don't filter by #service on relay side - some relays don't support it well
|
|
// Instead, fetch all kind 1 events since MIN_EVENT_DATE and filter client-side
|
|
const filters = [
|
|
{
|
|
kinds: [1],
|
|
since: MIN_EVENT_DATE, // January 5, 2026 00:00:00 UTC
|
|
limit: 1000, // Get up to 1000 events per sync
|
|
},
|
|
]
|
|
|
|
console.warn(`[PlatformSync] Starting sync with filter (no #service filter on relay side):`, JSON.stringify(filters, null, 2))
|
|
console.warn(`[PlatformSync] MIN_EVENT_DATE: ${MIN_EVENT_DATE} (${new Date(MIN_EVENT_DATE * 1000).toISOString()})`)
|
|
console.warn(`[PlatformSync] Will filter by service='${PLATFORM_SERVICE}' client-side after receiving events`)
|
|
|
|
const { relaySessionManager } = await import('./relaySessionManager')
|
|
const { syncProgressManager } = await import('./syncProgressManager')
|
|
const activeRelays = await relaySessionManager.getActiveRelays()
|
|
|
|
if (activeRelays.length === 0) {
|
|
throw new Error('No active relays available')
|
|
}
|
|
|
|
const allEvents: Event[] = []
|
|
const processedEventIds = new Set<string>()
|
|
|
|
// Synchronize from all active relays
|
|
for (let i = 0; i < activeRelays.length; i++) {
|
|
const relayUrl = activeRelays[i]
|
|
if (relayUrl) {
|
|
// Update progress with current relay
|
|
syncProgressManager.setProgress({
|
|
currentStep: 0,
|
|
totalSteps: activeRelays.length,
|
|
completed: false,
|
|
currentRelay: relayUrl,
|
|
})
|
|
|
|
try {
|
|
console.warn(`[PlatformSync] Synchronizing from relay ${i + 1}/${activeRelays.length}: ${relayUrl}`)
|
|
|
|
const sub = createSubscription(pool, [relayUrl], filters)
|
|
|
|
const relayEvents: Event[] = []
|
|
let resolved = false
|
|
let eventCount = 0
|
|
|
|
const finalize = async (): Promise<void> => {
|
|
if (resolved) {
|
|
return
|
|
}
|
|
resolved = true
|
|
sub.unsub()
|
|
|
|
// Deduplicate events by ID before adding to allEvents
|
|
for (const event of relayEvents) {
|
|
if (!processedEventIds.has(event.id)) {
|
|
processedEventIds.add(event.id)
|
|
allEvents.push(event)
|
|
}
|
|
}
|
|
|
|
console.warn(`[PlatformSync] Relay ${relayUrl} completed: received ${eventCount} total events from relay, ${relayEvents.length} filtered with service='${PLATFORM_SERVICE}'`)
|
|
|
|
// Update lastSyncDate for this relay on successful sync
|
|
if (relayEvents.length > 0 || eventCount > 0) {
|
|
const { configStorage } = await import('./configStorage')
|
|
const config = await configStorage.getConfig()
|
|
const relayConfig = config.relays.find((r) => r.url === relayUrl)
|
|
if (relayConfig) {
|
|
await configStorage.updateRelay(relayConfig.id, { lastSyncDate: Date.now() })
|
|
}
|
|
}
|
|
}
|
|
await new Promise<void>((resolve) => {
|
|
sub.on('event', (event: Event): void => {
|
|
eventCount = this.handleRelaySyncEvent({
|
|
event,
|
|
relayUrl,
|
|
relayEvents,
|
|
eventCount,
|
|
})
|
|
})
|
|
|
|
sub.on('eose', (): void => {
|
|
console.warn(`[PlatformSync] Relay ${relayUrl} sent EOSE signal`)
|
|
void finalize()
|
|
resolve()
|
|
})
|
|
|
|
// Timeout after SYNC_TIMEOUT_MS
|
|
const timeoutId = setTimeout((): void => {
|
|
console.warn(`[PlatformSync] Relay ${relayUrl} timeout after ${this.SYNC_TIMEOUT_MS}ms`)
|
|
void finalize()
|
|
resolve()
|
|
}, this.SYNC_TIMEOUT_MS)
|
|
timeoutId.unref?.()
|
|
|
|
this.syncSubscription = sub
|
|
})
|
|
|
|
// Update progress after each relay
|
|
syncProgressManager.setProgress({
|
|
currentStep: i + 1,
|
|
totalSteps: activeRelays.length,
|
|
completed: false,
|
|
currentRelay: relayUrl,
|
|
})
|
|
} catch (error) {
|
|
const errorMessage = error instanceof Error ? error.message : String(error)
|
|
console.warn(`[PlatformSync] Relay ${relayUrl} failed: ${errorMessage}`)
|
|
// Mark relay as failed but continue with next relay
|
|
relaySessionManager.markRelayFailed(relayUrl)
|
|
// Update progress even on failure
|
|
syncProgressManager.setProgress({
|
|
currentStep: i + 1,
|
|
totalSteps: activeRelays.length,
|
|
completed: false,
|
|
currentRelay: relayUrl,
|
|
})
|
|
}
|
|
}
|
|
}
|
|
|
|
// Process all collected events
|
|
await this.processAndCacheEvents(allEvents)
|
|
console.warn(`[PlatformSync] Total events collected from all relays: ${allEvents.length}`)
|
|
|
|
this.lastSyncTime = Date.now()
|
|
}
|
|
|
|
/**
|
|
* Process events and cache them by type
|
|
*/
|
|
private async processAndCacheEvents(events: Event[]): Promise<void> {
|
|
for (const event of events) {
|
|
try {
|
|
await this.processEvent(event)
|
|
} catch (error) {
|
|
console.error('Error processing event:', error, event.id)
|
|
// Continue processing other events even if one fails
|
|
}
|
|
}
|
|
}
|
|
|
|
private handleRelaySyncEvent(params: {
|
|
event: Event
|
|
relayUrl: string
|
|
relayEvents: Event[]
|
|
eventCount: number
|
|
}): number {
|
|
const nextCount = params.eventCount + 1
|
|
this.logRelaySyncProgress({ relayUrl: params.relayUrl, eventCount: nextCount })
|
|
this.logTargetEventReceived({ relayUrl: params.relayUrl, event: params.event, eventCount: nextCount })
|
|
|
|
const tags = extractTagsFromEvent(params.event)
|
|
this.logTargetEventTags({ event: params.event, tags })
|
|
|
|
if (tags.service === PLATFORM_SERVICE) {
|
|
params.relayEvents.push(params.event)
|
|
this.logTargetEventAccepted(params.event)
|
|
} else {
|
|
this.logTargetEventRejected({ event: params.event, tags })
|
|
}
|
|
|
|
return nextCount
|
|
}
|
|
|
|
private logRelaySyncProgress(params: { relayUrl: string; eventCount: number }): void {
|
|
if (params.eventCount % 100 === 0) {
|
|
console.warn(`[PlatformSync] Received ${params.eventCount} events from relay ${params.relayUrl} (client-side filtering in progress)`)
|
|
}
|
|
}
|
|
|
|
private logTargetEventReceived(params: { relayUrl: string; event: Event; eventCount: number }): void {
|
|
if (params.event.id !== TARGET_EVENT_ID) {
|
|
return
|
|
}
|
|
console.warn(`[PlatformSync] ✅ Received target event from relay ${params.relayUrl} (event #${params.eventCount}):`, {
|
|
id: params.event.id,
|
|
created_at: params.event.created_at,
|
|
created_at_date: new Date(params.event.created_at * 1000).toISOString(),
|
|
pubkey: params.event.pubkey,
|
|
allTags: params.event.tags,
|
|
serviceTags: params.event.tags.filter((tag) => tag[0] === 'service'),
|
|
})
|
|
}
|
|
|
|
private logTargetEventTags(params: { event: Event; tags: ReturnType<typeof extractTagsFromEvent> }): void {
|
|
if (params.event.id !== TARGET_EVENT_ID) {
|
|
return
|
|
}
|
|
console.warn(`[PlatformSync] Extracted tags for target event:`, {
|
|
extractedTags: params.tags,
|
|
hasServiceTag: params.tags.service === PLATFORM_SERVICE,
|
|
serviceValue: params.tags.service,
|
|
expectedService: PLATFORM_SERVICE,
|
|
})
|
|
}
|
|
|
|
private logTargetEventAccepted(event: Event): void {
|
|
if (event.id === TARGET_EVENT_ID) {
|
|
console.warn(`[PlatformSync] Target event accepted and added to relayEvents`)
|
|
}
|
|
}
|
|
|
|
private logTargetEventRejected(params: { event: Event; tags: ReturnType<typeof extractTagsFromEvent> }): void {
|
|
if (params.event.id !== TARGET_EVENT_ID) {
|
|
return
|
|
}
|
|
console.warn(`[PlatformSync] Event ${params.event.id} rejected: service tag is "${params.tags.service}", expected "${PLATFORM_SERVICE}"`)
|
|
}
|
|
|
|
/**
|
|
* Process a single event and cache it
|
|
*/
|
|
private async processEvent(event: Event): Promise<void> {
|
|
const tags = extractTagsFromEvent(event)
|
|
const { writeObjectToCache } = await import('./helpers/writeObjectHelper')
|
|
|
|
logTargetEventDebug({ event, tags })
|
|
if (tags.hidden) {
|
|
logTargetEventSkipped({ event, tags })
|
|
return
|
|
}
|
|
|
|
if (await this.tryCacheAuthorEvent({ event, tags, writeObjectToCache })) {
|
|
return
|
|
}
|
|
if (await this.tryCacheSeriesEvent({ event, tags, writeObjectToCache })) {
|
|
return
|
|
}
|
|
if (await this.tryCachePublicationEvent({ event, tags, writeObjectToCache })) {
|
|
return
|
|
}
|
|
if (await this.tryCacheReviewEvent({ event, tags, writeObjectToCache })) {
|
|
return
|
|
}
|
|
await this.tryCacheZapReceiptEvent({ event, writeObjectToCache })
|
|
}
|
|
|
|
private async tryCacheAuthorEvent(params: {
|
|
event: Event
|
|
tags: ReturnType<typeof extractTagsFromEvent>
|
|
writeObjectToCache: typeof import('./helpers/writeObjectHelper').writeObjectToCache
|
|
}): Promise<boolean> {
|
|
if (params.tags.type !== 'author') {
|
|
return false
|
|
}
|
|
|
|
logTargetEventAttempt({ event: params.event, message: 'Attempting to parse target event as author presentation' })
|
|
const parsed = await parsePresentationEvent(params.event)
|
|
logTargetEventParsed({ event: params.event, parsed })
|
|
|
|
if (!parsed?.hash) {
|
|
logTargetEventNotCached({ event: params.event, parsed })
|
|
return true
|
|
}
|
|
|
|
await params.writeObjectToCache({
|
|
objectType: 'author',
|
|
hash: parsed.hash,
|
|
event: params.event,
|
|
parsed,
|
|
version: params.tags.version,
|
|
hidden: params.tags.hidden,
|
|
index: parsed.index,
|
|
})
|
|
logTargetEventCached({ event: params.event, hash: parsed.hash })
|
|
return true
|
|
}
|
|
|
|
private async tryCacheSeriesEvent(params: {
|
|
event: Event
|
|
tags: ReturnType<typeof extractTagsFromEvent>
|
|
writeObjectToCache: typeof import('./helpers/writeObjectHelper').writeObjectToCache
|
|
}): Promise<boolean> {
|
|
if (params.tags.type !== 'series') {
|
|
return false
|
|
}
|
|
const parsed = await parseSeriesFromEvent(params.event)
|
|
if (!parsed?.hash) {
|
|
return true
|
|
}
|
|
await params.writeObjectToCache({
|
|
objectType: 'series',
|
|
hash: parsed.hash,
|
|
event: params.event,
|
|
parsed,
|
|
version: params.tags.version,
|
|
hidden: params.tags.hidden,
|
|
index: parsed.index,
|
|
})
|
|
return true
|
|
}
|
|
|
|
private async tryCachePublicationEvent(params: {
|
|
event: Event
|
|
tags: ReturnType<typeof extractTagsFromEvent>
|
|
writeObjectToCache: typeof import('./helpers/writeObjectHelper').writeObjectToCache
|
|
}): Promise<boolean> {
|
|
if (params.tags.type !== 'publication') {
|
|
return false
|
|
}
|
|
const parsed = await parseArticleFromEvent(params.event)
|
|
if (!parsed?.hash) {
|
|
return true
|
|
}
|
|
await params.writeObjectToCache({
|
|
objectType: 'publication',
|
|
hash: parsed.hash,
|
|
event: params.event,
|
|
parsed,
|
|
version: params.tags.version,
|
|
hidden: params.tags.hidden,
|
|
index: parsed.index,
|
|
})
|
|
return true
|
|
}
|
|
|
|
private async tryCacheReviewEvent(params: {
|
|
event: Event
|
|
tags: ReturnType<typeof extractTagsFromEvent>
|
|
writeObjectToCache: typeof import('./helpers/writeObjectHelper').writeObjectToCache
|
|
}): Promise<boolean> {
|
|
if (params.tags.type !== 'quote') {
|
|
return false
|
|
}
|
|
const parsed = await parseReviewFromEvent(params.event)
|
|
if (!parsed?.hash) {
|
|
return true
|
|
}
|
|
await params.writeObjectToCache({
|
|
objectType: 'review',
|
|
hash: parsed.hash,
|
|
event: params.event,
|
|
parsed,
|
|
version: params.tags.version,
|
|
hidden: params.tags.hidden,
|
|
index: parsed.index,
|
|
})
|
|
return true
|
|
}
|
|
|
|
private async tryCacheZapReceiptEvent(params: {
|
|
event: Event
|
|
writeObjectToCache: typeof import('./helpers/writeObjectHelper').writeObjectToCache
|
|
}): Promise<boolean> {
|
|
if (params.event.kind !== 9735) {
|
|
return false
|
|
}
|
|
|
|
if (await this.tryCacheSponsoringZapReceipt(params)) {
|
|
return true
|
|
}
|
|
if (await this.tryCachePurchaseZapReceipt(params)) {
|
|
return true
|
|
}
|
|
if (await this.tryCacheReviewTipZapReceipt(params)) {
|
|
return true
|
|
}
|
|
return true
|
|
}
|
|
|
|
private async tryCacheSponsoringZapReceipt(params: {
|
|
event: Event
|
|
writeObjectToCache: typeof import('./helpers/writeObjectHelper').writeObjectToCache
|
|
}): Promise<boolean> {
|
|
const sponsoring = await parseSponsoringFromEvent(params.event)
|
|
if (!sponsoring?.hash) {
|
|
return false
|
|
}
|
|
await params.writeObjectToCache({
|
|
objectType: 'sponsoring',
|
|
hash: sponsoring.hash,
|
|
event: params.event,
|
|
parsed: sponsoring,
|
|
index: sponsoring.index,
|
|
})
|
|
return true
|
|
}
|
|
|
|
private async tryCachePurchaseZapReceipt(params: {
|
|
event: Event
|
|
writeObjectToCache: typeof import('./helpers/writeObjectHelper').writeObjectToCache
|
|
}): Promise<boolean> {
|
|
const purchase = await parsePurchaseFromEvent(params.event)
|
|
if (!purchase?.hash) {
|
|
return false
|
|
}
|
|
await params.writeObjectToCache({
|
|
objectType: 'purchase',
|
|
hash: purchase.hash,
|
|
event: params.event,
|
|
parsed: purchase,
|
|
index: purchase.index,
|
|
})
|
|
return true
|
|
}
|
|
|
|
private async tryCacheReviewTipZapReceipt(params: {
|
|
event: Event
|
|
writeObjectToCache: typeof import('./helpers/writeObjectHelper').writeObjectToCache
|
|
}): Promise<boolean> {
|
|
const reviewTip = await parseReviewTipFromEvent(params.event)
|
|
if (!reviewTip?.hash) {
|
|
return false
|
|
}
|
|
await params.writeObjectToCache({
|
|
objectType: 'review_tip',
|
|
hash: reviewTip.hash,
|
|
event: params.event,
|
|
parsed: reviewTip,
|
|
index: reviewTip.index,
|
|
})
|
|
return true
|
|
}
|
|
|
|
/**
|
|
* Start continuous sync (runs periodically)
|
|
* Can use Service Worker if available, otherwise falls back to setInterval
|
|
*/
|
|
async startContinuousSync(): Promise<void> {
|
|
// Try to use Service Worker for background sync
|
|
if (typeof window !== 'undefined') {
|
|
try {
|
|
const { swClient } = await import('./swClient')
|
|
const isReady = await swClient.isReady()
|
|
if (isReady) {
|
|
console.warn('[PlatformSync] Using Service Worker for background sync')
|
|
await swClient.startPlatformSync()
|
|
// Still start initial sync in main thread
|
|
void this.startSync()
|
|
return
|
|
}
|
|
} catch (error) {
|
|
console.warn('[PlatformSync] Service Worker not available, using setInterval:', error)
|
|
}
|
|
}
|
|
|
|
// Fallback to setInterval if Service Worker not available
|
|
// Start initial sync
|
|
void this.startSync()
|
|
|
|
// Schedule periodic syncs
|
|
setInterval(() => {
|
|
if (!this.syncInProgress) {
|
|
void this.startSync()
|
|
}
|
|
}, this.SYNC_INTERVAL_MS)
|
|
}
|
|
|
|
/**
|
|
* Stop sync
|
|
*/
|
|
async stopSync(): Promise<void> {
|
|
// Stop Service Worker sync if active
|
|
if (typeof window !== 'undefined') {
|
|
try {
|
|
const { swClient } = await import('./swClient')
|
|
const isReady = await swClient.isReady()
|
|
if (isReady) {
|
|
await swClient.stopPlatformSync()
|
|
}
|
|
} catch {
|
|
// Ignore errors
|
|
}
|
|
}
|
|
|
|
// Stop local sync
|
|
if (this.syncSubscription) {
|
|
this.syncSubscription.unsub()
|
|
this.syncSubscription = null
|
|
}
|
|
this.syncInProgress = false
|
|
}
|
|
|
|
/**
|
|
* Check if sync is in progress
|
|
*/
|
|
isSyncing(): boolean {
|
|
return this.syncInProgress
|
|
}
|
|
|
|
/**
|
|
* Get last sync time
|
|
*/
|
|
getLastSyncTime(): number {
|
|
return this.lastSyncTime
|
|
}
|
|
}
|
|
|
|
export const platformSyncService = new PlatformSyncService()
|
|
|
|
function isTargetDebugEvent(eventId: string): boolean {
|
|
return eventId === '527d83e0af20bf23c3e104974090ccc21536ece72c24eb784b3642890f63b763'
|
|
}
|
|
|
|
function logTargetEventDebug(params: { event: Event; tags: ReturnType<typeof extractTagsFromEvent> }): void {
|
|
if (!isTargetDebugEvent(params.event.id)) {
|
|
return
|
|
}
|
|
console.warn(`[PlatformSync] Processing target event:`, {
|
|
id: params.event.id,
|
|
type: params.tags.type,
|
|
hidden: params.tags.hidden,
|
|
service: params.tags.service,
|
|
version: params.tags.version,
|
|
})
|
|
}
|
|
|
|
function logTargetEventSkipped(params: { event: Event; tags: ReturnType<typeof extractTagsFromEvent> }): void {
|
|
if (!isTargetDebugEvent(params.event.id) || !params.tags.hidden) {
|
|
return
|
|
}
|
|
console.warn(`[PlatformSync] Target event skipped: hidden=${params.tags.hidden}`)
|
|
}
|
|
|
|
function logTargetEventAttempt(params: { event: Event; message: string }): void {
|
|
if (!isTargetDebugEvent(params.event.id)) {
|
|
return
|
|
}
|
|
console.warn(`[PlatformSync] ${params.message}`)
|
|
}
|
|
|
|
function logTargetEventParsed(params: { event: Event; parsed: unknown }): void {
|
|
if (!isTargetDebugEvent(params.event.id)) {
|
|
return
|
|
}
|
|
const parsedObj = params.parsed as { hash?: string } | null
|
|
console.warn(`[PlatformSync] parsePresentationEvent result for target event:`, {
|
|
parsed: parsedObj !== null,
|
|
hasHash: parsedObj?.hash !== undefined,
|
|
hash: parsedObj?.hash,
|
|
})
|
|
}
|
|
|
|
function logTargetEventCached(params: { event: Event; hash: string }): void {
|
|
if (!isTargetDebugEvent(params.event.id)) {
|
|
return
|
|
}
|
|
console.warn(`[PlatformSync] Target event cached successfully as author with hash:`, params.hash)
|
|
}
|
|
|
|
function logTargetEventNotCached(params: { event: Event; parsed: unknown }): void {
|
|
if (!isTargetDebugEvent(params.event.id)) {
|
|
return
|
|
}
|
|
const parsedObj = params.parsed as { hash?: string } | null
|
|
console.warn(`[PlatformSync] Target event NOT cached: parsed=${parsedObj !== null}, hasHash=${parsedObj?.hash !== undefined}`)
|
|
}
|