import { existsSync, rmSync } from 'fs';
import { getPdsForDid, downloadAndCacheSite, extractBlobCid, fetchSiteRecord } from './utils';
import { upsertSite, tryAcquireLock, releaseLock } from './db';
import { safeFetch } from './safe-fetch';
import { isRecord, validateRecord } from '../lexicon/types/place/wisp/fs';
import { Firehose } from '@atproto/sync';
import { IdResolver } from '@atproto/identity';

const CACHE_DIR = './cache/sites';

export class FirehoseWorker {
  private firehose: Firehose | null = null;
  private idResolver: IdResolver;
  private isShuttingDown = false;
  private lastEventTime = Date.now();

  constructor(
    private logger?: (msg: string, data?: Record<string, unknown>) => void,
  ) {
    this.idResolver = new IdResolver();
  }

  private log(msg: string, data?: Record<string, unknown>) {
    const log = this.logger || console.log;
    log(`[FirehoseWorker] ${msg}`, data || {});
  }

  start() {
    this.log('Starting firehose worker');
    this.connect();
  }

  stop() {
    this.log('Stopping firehose worker');
    this.isShuttingDown = true;

    if (this.firehose) {
      this.firehose.destroy();
      this.firehose = null;
    }
  }

  private connect() {
    if (this.isShuttingDown) return;

    this.log('Connecting to AT Protocol firehose');

    this.firehose = new Firehose({
      idResolver: this.idResolver,
      service: 'wss://bsky.network',
      filterCollections: ['place.wisp.fs'],
      handleEvent: async (evt: any) => {
        this.lastEventTime = Date.now();

        // Watch for write events
        if (evt.event === 'create' || evt.event === 'update') {
          const record = evt.record;

          // If the write is a valid place.wisp.fs record
          if (
            evt.collection === 'place.wisp.fs' &&
            isRecord(record) &&
            validateRecord(record).success
          ) {
            this.log('Received place.wisp.fs event', {
              did: evt.did,
              event: evt.event,
              rkey: evt.rkey,
            });

            try {
              await this.handleCreateOrUpdate(evt.did, evt.rkey, record, evt.cid?.toString());
            } catch (err) {
              this.log('Error handling event', {
                did: evt.did,
                event: evt.event,
                rkey: evt.rkey,
                error: err instanceof Error ? err.message : String(err),
              });
            }
          }
        } else if (evt.event === 'delete' && evt.collection === 'place.wisp.fs') {
          this.log('Received delete event', {
            did: evt.did,
            rkey: evt.rkey,
          });

          try {
            await this.handleDelete(evt.did, evt.rkey);
          } catch (err) {
            this.log('Error handling delete', {
              did: evt.did,
              rkey: evt.rkey,
              error: err instanceof Error ? err.message : String(err),
            });
          }
        }
      },
      onError: (err: any) => {
        this.log('Firehose error', {
          error: err instanceof Error ? err.message : String(err),
          stack: err instanceof Error ? err.stack : undefined,
          fullError: err,
        });
        console.error('Full firehose error:', err);
      },
    });

    this.firehose.start();
    this.log('Firehose started');
  }

  private async handleCreateOrUpdate(did: string, site: string, record: any, eventCid?: string) {
    this.log('Processing create/update', { did, site });

    // Record is already validated in handleEvent
    const fsRecord = record;

    const pdsEndpoint = await getPdsForDid(did);
    if (!pdsEndpoint) {
      this.log('Could not resolve PDS for DID', { did });
      return;
    }

    this.log('Resolved PDS', { did, pdsEndpoint });

    // Verify record exists on PDS and fetch its CID
    let verifiedCid: string;
    try {
      const result = await fetchSiteRecord(did, site);

      if (!result) {
        this.log('Record not found on PDS, skipping cache', { did, site });
        return;
      }

      verifiedCid = result.cid;

      // Verify event CID matches PDS CID (prevent cache poisoning)
      if (eventCid && eventCid !== verifiedCid) {
        this.log('CID mismatch detected - potential spoofed event', {
          did,
          site,
          eventCid,
          verifiedCid
        });
        return;
      }

      this.log('Record verified on PDS', { did, site, cid: verifiedCid });
    } catch (err) {
      this.log('Failed to verify record on PDS', {
        did,
        site,
        error: err instanceof Error ? err.message : String(err),
      });
      return;
    }

    // Cache the record with verified CID (uses atomic swap internally)
    // All instances cache locally for edge serving
    await downloadAndCacheSite(did, site, fsRecord, pdsEndpoint, verifiedCid);

    // Acquire distributed lock only for database write to prevent duplicate writes
    const lockKey = `db:upsert:${did}:${site}`;
    const lockAcquired = await tryAcquireLock(lockKey);

    if (!lockAcquired) {
      this.log('Another instance is writing to DB, skipping upsert', { did, site });
      this.log('Successfully processed create/update (cached locally)', { did, site });
      return;
    }

    try {
      // Upsert site to database (only one instance does this)
      await upsertSite(did, site, fsRecord.site);
      this.log('Successfully processed create/update (cached + DB updated)', { did, site });
    } finally {
      // Always release lock, even if DB write fails
      await releaseLock(lockKey);
    }
  }

  private async handleDelete(did: string, site: string) {
    this.log('Processing delete', { did, site });

    // All instances should delete their local cache (no lock needed)
    const pdsEndpoint = await getPdsForDid(did);
    if (!pdsEndpoint) {
      this.log('Could not resolve PDS for DID', { did });
      return;
    }

    // Verify record is actually deleted from PDS
    try {
      const recordUrl = `${pdsEndpoint}/xrpc/com.atproto.repo.getRecord?repo=${encodeURIComponent(did)}&collection=place.wisp.fs&rkey=${encodeURIComponent(site)}`;
      const recordRes = await safeFetch(recordUrl);

      if (recordRes.ok) {
        this.log('Record still exists on PDS, not deleting cache', {
          did,
          site,
        });
        return;
      }

      this.log('Verified record is deleted from PDS', {
        did,
        site,
        status: recordRes.status,
      });
    } catch (err) {
      this.log('Error verifying deletion on PDS', {
        did,
        site,
        error: err instanceof Error ? err.message : String(err),
      });
    }

    // Delete cache
    this.deleteCache(did, site);

    this.log('Successfully processed delete', { did, site });
  }

  private deleteCache(did: string, site: string) {
    const cacheDir = `${CACHE_DIR}/${did}/${site}`;

    if (!existsSync(cacheDir)) {
      this.log('Cache directory does not exist, nothing to delete', {
        did,
        site,
      });
      return;
    }

    try {
      rmSync(cacheDir, { recursive: true, force: true });
      this.log('Cache deleted', { did, site, path: cacheDir });
    } catch (err) {
      this.log('Failed to delete cache', {
        did,
        site,
        path: cacheDir,
        error: err instanceof Error ? err.message : String(err),
      });
    }
  }

  getHealth() {
    const isConnected = this.firehose !== null;
    const timeSinceLastEvent = Date.now() - this.lastEventTime;

    return {
      connected: isConnected,
      lastEventTime: this.lastEventTime,
      timeSinceLastEvent,
      healthy: isConnected && timeSinceLastEvent < 300000, // 5 minutes
    };
  }
}