story-research-zapwall/lib/platformSync.ts

/**
 * 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 { nostrService } from './nostr'
import { PLATFORM_SERVICE, MIN_EVENT_DATE } from './platformConfig'
import { buildTagFilter, extractTagsFromEvent } from './nostrTagSystem'
import { objectCache } from './objectCache'
import { parsePresentationEvent } from './articlePublisherHelpersPresentation'
import { parseArticleFromEvent, parseSeriesFromEvent, parseReviewFromEvent, parsePurchaseFromEvent, parseReviewTipFromEvent, parseSponsoringFromEvent } from './nostrEventParsing'

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 = 30000 // 30 seconds timeout per sync

  /**
   * 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...'
    syncProgressManager.setProgress({ currentStep: 0, totalSteps: 1, completed: false, currentRelay: initialRelay })

    try {
      await this.performSync(pool as unknown as SimplePoolWithSub)
      syncProgressManager.setProgress({ currentStep: 1, totalSteps: 1, completed: true, currentRelay: initialRelay })
    } 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
   */
  private async performSync(pool: SimplePoolWithSub): Promise<void> {
    const filters = [
      {
        ...buildTagFilter({
          service: PLATFORM_SERVICE,
        }),
        since: MIN_EVENT_DATE,
        limit: 1000, // Get up to 1000 events per sync
      },
    ]

    // Use relay rotation for platform sync
    const { tryWithRelayRotation } = await import('./relayRotation')
    const { syncProgressManager } = await import('./syncProgressManager')

    return tryWithRelayRotation(
      pool as unknown as import('nostr-tools').SimplePool,
      async (relayUrl, poolWithSub) => {
        // Update progress with current relay
        const currentProgress = syncProgressManager.getProgress()
        if (currentProgress) {
          syncProgressManager.setProgress({
            ...currentProgress,
            currentRelay: relayUrl,
          })
        }

        const { createSubscription } = require('@/types/nostr-tools-extended')
        const sub = createSubscription(poolWithSub, [relayUrl], filters)

        const events: Event[] = []
        let resolved = false

        const finalize = async (): Promise<void> => {
          if (resolved) {
            return
          }
          resolved = true
          sub.unsub()
          this.syncSubscription = null

          // Process all events and cache them
          await this.processAndCacheEvents(events)

          this.lastSyncTime = Date.now()
        }

        return new Promise<void>((resolve) => {
          sub.on('event', (event: Event): void => {
            // Only process events with service='zapwall.fr'
            const tags = extractTagsFromEvent(event)
            if (tags.service === PLATFORM_SERVICE) {
              events.push(event)
            }
          })

          sub.on('eose', (): void => {
            void (async (): Promise<void> => {
              await finalize()
              resolve()
            })()
          })

          // Timeout after SYNC_TIMEOUT_MS
          setTimeout((): void => {
            void (async (): Promise<void> => {
              await finalize()
              resolve()
            })()
          }, this.SYNC_TIMEOUT_MS).unref?.()

          this.syncSubscription = sub
        })
      },
      30000 // 30 second timeout per relay
    )
  }

  /**
   * 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
      }
    }
  }

  /**
   * Process a single event and cache it
   */
  private async processEvent(event: Event): Promise<void> {
    const tags = extractTagsFromEvent(event)

    // Skip hidden events
    if (tags.hidden) {
      return
    }

    // Try to parse and cache by type
    if (tags.type === 'author') {
      const parsed = await parsePresentationEvent(event)
      if (parsed && parsed.hash) {
        await objectCache.set('author', parsed.hash, event, parsed, tags.version ?? 0, tags.hidden, parsed.index)
      }
    } else if (tags.type === 'series') {
      const parsed = await parseSeriesFromEvent(event)
      if (parsed && parsed.hash) {
        await objectCache.set('series', parsed.hash, event, parsed, tags.version ?? 0, tags.hidden, parsed.index)
      }
    } else if (tags.type === 'publication') {
      const parsed = await parseArticleFromEvent(event)
      if (parsed && parsed.hash) {
        await objectCache.set('publication', parsed.hash, event, parsed, tags.version ?? 0, tags.hidden, parsed.index)
      }
    } else if (tags.type === 'quote') {
      const parsed = await parseReviewFromEvent(event)
      if (parsed && parsed.hash) {
        await objectCache.set('review', parsed.hash, event, parsed, tags.version ?? 0, tags.hidden, parsed.index)
      }
    } else if (event.kind === 9735) {
      // Zap receipts (kind 9735) can be sponsoring, purchase, or review_tip
      const sponsoring = await parseSponsoringFromEvent(event)
      if (sponsoring && sponsoring.hash) {
        await objectCache.set('sponsoring', sponsoring.hash, event, sponsoring, 0, false, sponsoring.index)
      } else {
        const purchase = await parsePurchaseFromEvent(event)
        if (purchase && purchase.hash) {
          await objectCache.set('purchase', purchase.hash, event, purchase, 0, false, purchase.index)
        } else {
          const reviewTip = await parseReviewTipFromEvent(event)
          if (reviewTip && reviewTip.hash) {
            await objectCache.set('review_tip', reviewTip.hash, event, reviewTip, 0, false, reviewTip.index)
          }
        }
      }
    }
  }

  /**
   * Start continuous sync (runs periodically)
   */
  startContinuousSync(): void {
    // Start initial sync
    void this.startSync()

    // Schedule periodic syncs
    setInterval(() => {
      if (!this.syncInProgress) {
        void this.startSync()
      }
    }, this.SYNC_INTERVAL_MS)
  }

  /**
   * Stop sync
   */
  stopSync(): void {
    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()