import { readFileSync, writeFileSync } from 'fs'
import WebSocket from 'ws'
import { DynamoDBClient } from '@aws-sdk/client-dynamodb'
import { DynamoDBDocumentClient, GetCommand, PutCommand, DeleteCommand, UpdateCommand, QueryCommand } from '@aws-sdk/lib-dynamodb'

const TABLE = process.env.DYNAMODB_TABLE ?? 'blogs'
const JETSTREAM_HOST = process.env.JETSTREAM_HOST ?? 'jetstream2.us-east.bsky.network'
const CURSOR_FILE = process.env.CURSOR_FILE ?? './cursor'
const ACCEPT_LOOSE_DOCUMENTS = process.env.ACCEPT_LOOSE_DOCUMENTS === 'true'
const DEBUG = process.env.NODE_ENV !== 'production'
const CELEBRITY_THRESHOLD = parseInt(process.env.CELEBRITY_THRESHOLD ?? '1000', 10)
const TIMELINE_TTL_SECONDS = 30 * 24 * 60 * 60

const COLLECTIONS = [
  'site.standard.publication',
  'site.standard.document',
  'site.standard.graph.subscription',
  'pub.lemma.blog.draft',
]

const ddb = DynamoDBDocumentClient.from(
  new DynamoDBClient(
    process.env.AWS_ENDPOINT_URL
      ? { endpoint: process.env.AWS_ENDPOINT_URL, region: process.env.AWS_REGION }
      : {},
  ),
)

// ── Cursor ────────────────────────────────────────────────────────────────────
// Jetstream time_us cursor lets us resume from the last processed event after a
// restart rather than reprocessing everything or missing events.

let cursor: number | undefined

function loadCursor(): void {
  try {
    const raw = readFileSync(CURSOR_FILE, 'utf8').trim()
    const n = parseInt(raw, 10)
    if (!isNaN(n)) cursor = n
  } catch {
    // No cursor file yet — start from now.
  }
}

function saveCursor(timeUs: number): void {
  cursor = timeUs
  try {
    writeFileSync(CURSOR_FILE, String(timeUs))
  } catch (err) {
    console.error('Failed to save cursor:', err)
  }
}

// ── Publication URL resolution ────────────────────────────────────────────────

type DidDoc = {
  alsoKnownAs?: string[]
  service?: { type: string; serviceEndpoint: string }[]
}

async function resolveDidDoc(did: string): Promise<DidDoc> {
  const docUrl = did.startsWith('did:plc:')
    ? `https://plc.directory/${did}`
    : did.startsWith('did:web:')
      ? `https://${did.slice('did:web:'.length)}/.well-known/did.json`
      : (() => { throw new Error(`Unsupported DID method: ${did}`) })()
  const res = await fetch(docUrl)
  if (!res.ok) throw new Error(`Failed to resolve DID ${did}: ${res.status}`)
  return res.json() as Promise<DidDoc>
}

async function resolvePdsUrl(did: string): Promise<string> {
  const doc = await resolveDidDoc(did)
  const pds = doc.service?.find((s) => s.type === 'AtprotoPersonalDataServer')?.serviceEndpoint
  if (!pds) throw new Error(`No PDS service endpoint for ${did}`)
  return pds
}

async function resolveAuthorHandle(did: string): Promise<string> {
  try {
    const doc = await resolveDidDoc(did)
    return doc.alsoKnownAs?.[0]?.replace('at://', '') ?? did
  } catch {
    return did
  }
}

async function resolvePublicationMeta(atUri: string): Promise<{ url: string; name: string }> {
  const parts = atUri.slice('at://'.length).split('/')
  const pubDid = parts[0]
  const pubRkey = parts[2]

  const cached = await ddb.send(new GetCommand({
    TableName: TABLE,
    Key: { PK: `u#${pubDid}#pub`, SK: `pub#${pubRkey}` },
  }))
  if (cached.Item?.url) {
    return {
      url: (cached.Item.url as string).replace(/\/$/, ''),
      name: (cached.Item.name as string | undefined) ?? '',
    }
  }

  const pds = await resolvePdsUrl(pubDid)
  const res = await fetch(
    `${pds}/xrpc/com.atproto.repo.getRecord?repo=${encodeURIComponent(pubDid)}&collection=site.standard.publication&rkey=${encodeURIComponent(pubRkey)}`,
  )
  if (res.status === 404) throw new Error(`Publication not found: ${atUri}`)
  if (!res.ok) throw new Error(`Publication fetch failed (${res.status}): ${atUri}`)
  const record = (await res.json()) as {
    value: { url: string; name: string; description?: string; preferences?: { showInDiscover?: boolean } }
  }

  // Cache the publication so future lookups are fast.
  await upsertPublication(pubDid, pubRkey, record.value)

  return { url: record.value.url.replace(/\/$/, ''), name: record.value.name ?? '' }
}

// ── DynamoDB writes ───────────────────────────────────────────────────────────

type PublicationRecord = {
  url: string
  name: string
  description?: string
  preferences?: { showInDiscover?: boolean }
}

type DocumentRecord = {
  site: string
  path?: string
  title: string
  description?: string
  publishedAt: string
  updatedAt?: string
  textContent?: string
  content?: { $type?: string; content?: string; license?: string }
  coverImage?: { ref: { $link: string } }
  bskyPostRef?: { uri: string; cid: string }
}

type SubscriptionRecord = {
  publication: string
  createdAt?: string
}

async function upsertPublication(did: string, rkey: string, record: PublicationRecord) {
  const atUri = `at://${did}/site.standard.publication/${rkey}`
  const normalizedUrl = decodeURIComponent(record.url.replace(/\/$/, ''))
  const resolvedHandle = await resolveAuthorHandle(did)
  const item: Record<string, unknown> = {
    PK: `u#${did}#pub`,
    SK: `pub#${rkey}`,
    entityType: 'publication',
    atUri,
    url: normalizedUrl,
    name: record.name,
    resolvedHandle,
    preferences: { showInDiscover: record.preferences?.showInDiscover ?? false },
  }
  if (record.description?.trim()) item.description = record.description.trim()
  await Promise.all([
    ddb.send(new PutCommand({ TableName: TABLE, Item: item })),
    ddb.send(new PutCommand({
      TableName: TABLE,
      Item: { PK: `url#${normalizedUrl}`, SK: 'info', entityType: 'url-lookup', pubAtUri: atUri },
    })),
  ])
}

async function deletePublication(did: string, rkey: string) {
  const key = { PK: `u#${did}#pub`, SK: `pub#${rkey}` }
  const existing = await ddb.send(new GetCommand({ TableName: TABLE, Key: key }))
  const pubUrl = existing.Item?.url as string | undefined
  const deletes: Promise<unknown>[] = [
    ddb.send(new DeleteCommand({ TableName: TABLE, Key: key })),
  ]
  if (pubUrl) {
    deletes.push(ddb.send(new DeleteCommand({
      TableName: TABLE,
      Key: { PK: `url#${pubUrl.replace(/\/$/, '')}`, SK: 'info' },
    })))
  }
  await Promise.all(deletes)
}

async function removeAuthorFromTimeline(readerDid: string, authorDid: string): Promise<void> {
  let lastKey: Record<string, unknown> | undefined
  do {
    const result = await ddb.send(new QueryCommand({
      TableName: TABLE,
      KeyConditionExpression: 'PK = :pk',
      FilterExpression: 'authorDid = :authorDid',
      ExpressionAttributeValues: { ':pk': `u#${readerDid}#timeline`, ':authorDid': authorDid },
      ProjectionExpression: 'PK, SK',
      ExclusiveStartKey: lastKey,
    }))
    await Promise.all((result.Items ?? []).map((item) =>
      ddb.send(new DeleteCommand({ TableName: TABLE, Key: { PK: item.PK, SK: item.SK } }))
    ))
    lastKey = result.LastEvaluatedKey as Record<string, unknown> | undefined
  } while (lastKey)
}

async function handleSubscription(
  readerDid: string,
  rkey: string,
  record: SubscriptionRecord | undefined,
  operation: 'create' | 'update' | 'delete',
): Promise<void> {
  if (operation === 'delete') {
    const existing = await ddb.send(new GetCommand({
      TableName: TABLE,
      Key: { PK: `u#${readerDid}#subscriptions`, SK: `sub#${rkey}` },
    }))
    if (!existing.Item) return
    const authorDid = existing.Item.authorDid as string
    await Promise.all([
      ddb.send(new DeleteCommand({
        TableName: TABLE,
        Key: { PK: `u#${readerDid}#subscriptions`, SK: `sub#${rkey}` },
      })),
      ddb.send(new DeleteCommand({
        TableName: TABLE,
        Key: { PK: `u#${authorDid}#subscribers`, SK: `u#${readerDid}` },
      })),
      ddb.send(new UpdateCommand({
        TableName: TABLE,
        Key: { PK: `u#${authorDid}`, SK: 'count' },
        UpdateExpression: 'ADD subscribers :dec',
        ExpressionAttributeValues: { ':dec': -1 },
      })),
      removeAuthorFromTimeline(readerDid, authorDid),
    ])
    console.log(`del  sub  ${readerDid}  → ${authorDid}`)
    return
  }

  if (!record) return
  const pubAtUri = record.publication
  if (!pubAtUri?.startsWith('at://')) {
    console.warn(`sub  skip  ${readerDid}  invalid publication: ${pubAtUri}`)
    return
  }
  const authorDid = pubAtUri.slice('at://'.length).split('/')[0]
  const subscribedAt = record.createdAt ?? new Date().toISOString()

  await Promise.all([
    ddb.send(new PutCommand({
      TableName: TABLE,
      Item: {
        PK: `u#${readerDid}#subscriptions`,
        SK: `sub#${rkey}`,
        entityType: 'subscription',
        publicationAtUri: pubAtUri,
        authorDid,
        subscribedAt,
      },
    })),
    ddb.send(new PutCommand({
      TableName: TABLE,
      Item: {
        PK: `u#${authorDid}#subscribers`,
        SK: `u#${readerDid}`,
        entityType: 'subscriber',
        subscribedAt,
      },
    })),
    ddb.send(new UpdateCommand({
      TableName: TABLE,
      Key: { PK: `u#${authorDid}`, SK: 'count' },
      UpdateExpression: 'ADD subscribers :inc',
      ExpressionAttributeValues: { ':inc': 1 },
    })),
  ])
  await fanInExistingPosts(readerDid, authorDid, pubAtUri)
  console.log(`${operation === 'create' ? 'new' : 'upd'}  sub  ${readerDid}  → ${authorDid}`)
}

async function fanOutToTimeline(
  authorDid: string,
  rkey: string,
  record: DocumentRecord,
  pubMeta: { url: string; name: string },
  resolvedHandle: string,
  pubAtUri: string,
): Promise<void> {
  const countItem = await ddb.send(new GetCommand({
    TableName: TABLE,
    Key: { PK: `u#${authorDid}`, SK: 'count' },
  }))
  const subscriberCount = (countItem.Item?.subscribers as number | undefined) ?? 0
  if (subscriberCount > CELEBRITY_THRESHOLD) {
    if (DEBUG) console.log(`  fanout skip ${authorDid} (celebrity: ${subscriberCount})`)
    return
  }
  if (subscriberCount === 0) return

  const ttl = Math.floor(Date.now() / 1000) + TIMELINE_TTL_SECONDS
  let lastKey: Record<string, unknown> | undefined

  do {
    const res = await ddb.send(new QueryCommand({
      TableName: TABLE,
      KeyConditionExpression: 'PK = :pk',
      ExpressionAttributeValues: { ':pk': `u#${authorDid}#subscribers` },
      ExclusiveStartKey: lastKey,
    }))

    const writes = (res.Items ?? []).map((subscriber) => {
      const readerDid = (subscriber.SK as string).slice('u#'.length)
      const item: Record<string, unknown> = {
        PK: `u#${readerDid}#timeline`,
        SK: `${new Date(Math.min(new Date(record.publishedAt).getTime(), Date.now())).toISOString()}#${rkey}`,
        entityType: 'timeline-item',
        authorDid,
        rkey,
        title: record.title,
        publishedAt: record.publishedAt,
        publicationName: pubMeta.name,
        pubAtUri,
        resolvedUrl: pubMeta.url,
        path: record.path ?? `/${rkey}`,
        resolvedHandle,
        ttl,
      }
      if (record.coverImage?.ref.$link) item.coverImageCid = record.coverImage.ref.$link
      return ddb.send(new PutCommand({ TableName: TABLE, Item: item }))
    })

    await Promise.all(writes)
    lastKey = res.LastEvaluatedKey as Record<string, unknown> | undefined
  } while (lastKey)

  if (DEBUG) console.log(`  fanout ${authorDid} → ${subscriberCount} subscriber(s)`)
}

async function fanInExistingPosts(readerDid: string, authorDid: string, publicationAtUri: string): Promise<void> {
  const posts: Record<string, unknown>[] = []
  let lastKey: Record<string, unknown> | undefined

  do {
    const result = await ddb.send(new QueryCommand({
      TableName: TABLE,
      KeyConditionExpression: 'PK = :pk',
      FilterExpression: 'visibility = :vis AND site = :site',
      ExpressionAttributeValues: { ':pk': `u#${authorDid}#post`, ':vis': 'published', ':site': publicationAtUri },
      ScanIndexForward: false,
      Limit: 100,
      ExclusiveStartKey: lastKey,
    }))
    posts.push(...(result.Items ?? []))
    lastKey = result.LastEvaluatedKey as Record<string, unknown> | undefined
  } while (lastKey && posts.length < 20)

  const recentPosts = posts.slice(0, 20)
  if (recentPosts.length === 0) return

  const ttl = Math.floor(Date.now() / 1000) + TIMELINE_TTL_SECONDS
  await Promise.all(recentPosts.map((post) => {
    const postRkey = (post.SK as string).slice('p#'.length)
    const item: Record<string, unknown> = {
      PK: `u#${readerDid}#timeline`,
      SK: `${(post.GSI1SK as string | undefined)?.split('#')[0] ?? new Date(Math.min(new Date(post.publishedAt as string).getTime(), Date.now())).toISOString()}#${postRkey}`,
      entityType: 'timeline-item',
      authorDid,
      rkey: postRkey,
      title: post.title as string,
      publishedAt: post.publishedAt as string,
      publicationName: post.resolvedPubName as string | undefined,
      pubAtUri: typeof post.site === 'string' && post.site.startsWith('at://') ? post.site : undefined,
      resolvedUrl: post.resolvedUrl as string | undefined,
      path: post.path as string | undefined,
      resolvedHandle: post.resolvedHandle as string | undefined,
      ttl,
    }
    if (post.coverImageRef) item.coverImageCid = post.coverImageRef as string
    return ddb.send(new PutCommand({ TableName: TABLE, Item: item }))
  }))
}

async function upsertDocument(did: string, rkey: string, record: DocumentRecord) {
  const site = record.site
  const siteIsAtUri = site.startsWith('at://')

  let pubMeta: { url: string; name: string }
  let resolvedHandle: string
  try {
    ;[pubMeta, resolvedHandle] = await Promise.all([
      siteIsAtUri
        ? resolvePublicationMeta(site)
        : Promise.resolve({ url: site.replace(/\/$/, ''), name: '' }),
      resolveAuthorHandle(did),
    ])
  } catch (err) {
    if (err instanceof Error && err.message.startsWith('Publication not found:')) {
      if (DEBUG) console.log(`skip doc  ${did}  ${rkey}  (pub not found on PDS)`)
      return
    }
    throw err
  }

  const existing = await ddb.send(new GetCommand({
    TableName: TABLE,
    Key: { PK: `u#${did}#post`, SK: `p#${rkey}` },
    ProjectionExpression: 'GSI1SK',
  }))
  const gsi1sk = (existing.Item?.GSI1SK as string | undefined)
    ?? `${new Date(Math.min(new Date(record.publishedAt).getTime(), Date.now())).toISOString()}#${rkey}`

  const item: Record<string, unknown> = {
    PK: `u#${did}#post`,
    SK: `p#${rkey}`,
    GSI1PK: 'blog-post#published',
    GSI1SK: gsi1sk,
    entityType: 'blog-post',
    visibility: 'published',
    atUri: `at://${did}/site.standard.document/${rkey}`,
    site,
    resolvedUrl: pubMeta.url,
    resolvedPubName: pubMeta.name,
    resolvedHandle,
    path: record.path ?? `/${rkey}`,
    title: record.title,
    publishedAt: record.publishedAt,
  }
  if (record.updatedAt) item.updatedAt = record.updatedAt
  if (record.description?.trim()) item.description = record.description.trim()
  const contentEntry = record.content
  if (contentEntry?.$type === 'pub.lemma.blog.entry' && contentEntry.content !== undefined) {
    item.content = contentEntry.content
    if (contentEntry.license) item.license = contentEntry.license
  }
  if (record.textContent !== undefined) item.textContent = record.textContent
  if (record.coverImage?.ref.$link) item.coverImageRef = record.coverImage.ref.$link
  if (record.bskyPostRef) {
    const refDid = (record.bskyPostRef as { uri?: string }).uri?.split('/')[2]
    if (refDid === did) item.bskyPostRef = record.bskyPostRef
  }
  await ddb.send(new PutCommand({ TableName: TABLE, Item: item }))

  await fanOutToTimeline(did, rkey, record, pubMeta, resolvedHandle, siteIsAtUri ? site : '')
}

async function deleteDocument(did: string, rkey: string) {
  await ddb.send(new DeleteCommand({
    TableName: TABLE,
    Key: { PK: `u#${did}#post`, SK: `p#${rkey}` },
  }))
}

async function upsertDraft(did: string, rkey: string, record: DocumentRecord) {
  const item: Record<string, unknown> = {
    PK: `u#${did}#post`,
    SK: `draft#${rkey}`,
    entityType: 'blog-draft',
    visibility: 'draft',
    atUri: `at://${did}/pub.lemma.blog.draft/${rkey}`,
    site: record.site,
    title: record.title,
    publishedAt: record.publishedAt,
  }
  if (record.updatedAt) item.updatedAt = record.updatedAt
  if (record.coverImage?.ref.$link) item.coverImageRef = record.coverImage.ref.$link
  await ddb.send(new PutCommand({ TableName: TABLE, Item: item }))
}

async function deleteDraft(did: string, rkey: string) {
  await ddb.send(new DeleteCommand({
    TableName: TABLE,
    Key: { PK: `u#${did}#post`, SK: `draft#${rkey}` },
  }))
}

// ── Event handling ────────────────────────────────────────────────────────────

type JetstreamEvent = {
  did: string
  time_us: number
  kind: string
  commit?: {
    operation: 'create' | 'update' | 'delete'
    collection: string
    rkey: string
    record?: Record<string, unknown>
  }
}

// ── Muted users ───────────────────────────────────────────────────────────────

const mutedUsers = new Set<string>()

async function loadMutedUsers(): Promise<void> {
  let lastKey: Record<string, unknown> | undefined
  do {
    const result = await ddb.send(new QueryCommand({
      TableName: TABLE,
      KeyConditionExpression: 'PK = :pk',
      ExpressionAttributeValues: { ':pk': 'muted#user' },
      ProjectionExpression: 'SK',
      ExclusiveStartKey: lastKey,
    }))
    for (const item of result.Items ?? []) {
      const did = (item.SK as string).slice('u#'.length)
      mutedUsers.add(did)
    }
    lastKey = result.LastEvaluatedKey as Record<string, unknown> | undefined
  } while (lastKey)
  console.log(`Loaded ${mutedUsers.size} muted user(s)`)
}

async function handleEvent(event: JetstreamEvent): Promise<void> {
  if (event.kind !== 'commit' || !event.commit) return
  const { did, commit } = event
  const { operation, collection, rkey, record } = commit

  if (collection === 'site.standard.publication') {
    if (operation === 'delete') {
      await deletePublication(did, rkey)
      console.log(`del  pub  ${did}  ${rkey}`)
    } else if (record) {
      if (mutedUsers.has(did)) {
        if (DEBUG) console.log(`skip pub  ${did}  ${rkey}  (muted)`)
        return
      }
      await upsertPublication(did, rkey, record as unknown as PublicationRecord)
      console.log(`${operation === 'create' ? 'new' : 'upd'}  pub  ${did}  ${rkey}`)
    }
  } else if (collection === 'site.standard.document') {
    if (operation === 'delete') {
      await deleteDocument(did, rkey)
      console.log(`del  doc  ${did}  ${rkey}`)
    } else if (record) {
      if (mutedUsers.has(did)) {
        if (DEBUG) console.log(`skip doc  ${did}  ${rkey}  (muted)`)
        return
      }
      const doc = record as unknown as DocumentRecord
      if (!doc.site.startsWith('at://') && !ACCEPT_LOOSE_DOCUMENTS) {
        if (DEBUG) console.log(`skip doc  ${did}  ${rkey}  (loose document)`)
        return
      }
      await upsertDocument(did, rkey, doc)
      console.log(`${operation === 'create' ? 'new' : 'upd'}  doc  ${did}  ${rkey}  "${doc.title}"`)
    }
  } else if (collection === 'site.standard.graph.subscription') {
    await handleSubscription(did, rkey, record as unknown as SubscriptionRecord | undefined, operation)
  } else if (collection === 'pub.lemma.blog.draft') {
    if (operation === 'delete') {
      await deleteDraft(did, rkey)
      console.log(`del  draft  ${did}  ${rkey}`)
    } else if (record) {
      if (mutedUsers.has(did)) {
        if (DEBUG) console.log(`skip draft  ${did}  ${rkey}  (muted)`)
        return
      }
      const doc = record as unknown as DocumentRecord
      await upsertDraft(did, rkey, doc)
      console.log(`${operation === 'create' ? 'new' : 'upd'}  draft  ${did}  ${rkey}  "${doc.title}"`)
    }
  }
}

// ── WebSocket connection with reconnection ────────────────────────────────────

let stopping = false
let reconnectDelay = 1000

function buildUrl(): string {
  const params = new URLSearchParams()
  for (const c of COLLECTIONS) params.append('wantedCollections', c)
  if (cursor !== undefined) params.set('cursor', String(cursor))
  return `wss://${JETSTREAM_HOST}/subscribe?${params}`
}

function connect(): void {
  const url = buildUrl()
  console.log(`Connecting to ${url}`)
  const ws = new WebSocket(url)

  ws.on('open', () => {
    console.log('Connected')
    reconnectDelay = 1000
  })

  ws.on('message', (data) => {
    let event: JetstreamEvent
    try {
      event = JSON.parse(data.toString()) as JetstreamEvent
    } catch {
      return
    }
    saveCursor(event.time_us)
    handleEvent(event).catch((err) => console.error('Handler error:', err))
  })

  ws.on('close', () => {
    if (stopping) return
    console.log(`Disconnected — reconnecting in ${reconnectDelay / 1000}s`)
    setTimeout(connect, reconnectDelay)
    reconnectDelay = Math.min(reconnectDelay * 2, 30_000)
  })

  ws.on('error', (err) => {
    console.error('WebSocket error:', err.message)
    ws.terminate()
  })

  process.on('SIGINT', () => {
    stopping = true
    console.log('Shutting down')
    ws.close()
    process.exit(0)
  })
}

// ── Entry point ───────────────────────────────────────────────────────────────

loadCursor()
loadMutedUsers().then(connect)
