// Lambda handler: triggered by s3:ObjectCreated on the backfill trigger bucket.
// CLI usage: node dist/index.js <dids-file>
// Reads one DID per line (blank lines and # comments ignored).
// Pass 1: upserts all site.standard.publication and site.standard.document records.
// Pass 2: upserts all site.standard.graph.subscription records and fans in existing posts.
// Processing publications and documents before subscriptions ensures fan-in finds indexed posts.

import { readFileSync } from 'fs'
import { DynamoDBClient } from '@aws-sdk/client-dynamodb'
import {
  DynamoDBDocumentClient,
  GetCommand,
  PutCommand,
  UpdateCommand,
  QueryCommand,
} from '@aws-sdk/lib-dynamodb'
import { S3Client, GetObjectCommand } from '@aws-sdk/client-s3'

const TABLE = process.env.DYNAMODB_TABLE ?? 'blogs'
const CONCURRENCY = parseInt(process.env.CONCURRENCY ?? '5', 10)

function markdownToText(md: string): string {
  return md
    .replace(/```[\s\S]*?```/g, '')
    .replace(/`[^`]*`/g, '')
    .replace(/!\[.*?\]\(.*?\)/g, '')
    .replace(/\[([^\]]+)\]\([^)]+\)/g, '$1')
    .replace(/^#{1,6}\s+/gm, '')
    .replace(/(\*\*|__)(.*?)\1/g, '$2')
    .replace(/(\*|_)(.*?)\1/g, '$2')
    .replace(/^[-*+]\s+/gm, '')
    .replace(/^\d+\.\s+/gm, '')
    .replace(/^>\s+/gm, '')
    .replace(/\n{2,}/g, ' ')
    .replace(/\s+/g, ' ')
    .trim()
}
const ACCEPT_LOOSE_DOCUMENTS = process.env.ACCEPT_LOOSE_DOCUMENTS === 'true'
const TIMELINE_TTL_SECONDS = 30 * 24 * 60 * 60

function endpointConfig(extra?: { forcePathStyle?: boolean }) {
  if (!process.env.AWS_ENDPOINT_URL) return {}
  return {
    endpoint: process.env.AWS_ENDPOINT_URL,
    region: process.env.AWS_REGION,
    ...extra,
  }
}

const ddb = DynamoDBDocumentClient.from(new DynamoDBClient(endpointConfig()))

const s3 = new S3Client(endpointConfig({ forcePathStyle: true }))

// ── DID / PDS resolution ──────────────────────────────────────────────────────

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

async function resolveDidDoc(did: string): Promise<DidDoc> {
  const url = 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(url)
  if (!res.ok) throw new Error(`Failed to resolve DID doc for ${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 endpoint for ${did}`)
  return pds
}

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

// ── Publication metadata ──────────────────────────────────────────────────────

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.ok) throw new Error(`Publication not found: ${atUri}`)
  const { value } = (await res.json()) as {
    value: { url: string; name: string; description?: string; preferences?: { showInDiscover?: boolean } }
  }

  // Cache it so subsequent documents from the same publication don't re-fetch.
  const handle = await resolveHandle(pubDid)
  await upsertPublication(pubDid, pubRkey, value, handle)

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

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

async function upsertPublication(
  did: string,
  rkey: string,
  record: { url: string; name: string; description?: string; preferences?: { showInDiscover?: boolean } },
  resolvedHandle: string,
): Promise<void> {
  const atUri = `at://${did}/site.standard.publication/${rkey}`
  const normalizedUrl = decodeURIComponent(record.url.replace(/\/$/, ''))
  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 upsertDocument(
  did: string,
  rkey: string,
  record: {
    site: string
    path?: string
    title: string
    description?: string
    publishedAt: string
    updatedAt?: string
    content?: { $type?: string; content?: string }
    coverImage?: { ref: { $link: string } }
    bskyPostRef?: { uri: string; cid: string }
  },
  resolvedHandle: string,
): Promise<void> {
  const siteIsAtUri = record.site.startsWith('at://')
  const pubMeta = siteIsAtUri
    ? await resolvePublicationMeta(record.site)
    : { url: record.site.replace(/\/$/, ''), name: '' }

  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: record.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()
  if (record.content?.$type === 'pub.lemma.blog.entry' && record.content.content !== undefined) {
    item.content = record.content.content
    const text = markdownToText(record.content.content)
    if (text) item.textContent = text
  }
  if (record.coverImage?.ref.$link) item.coverImageRef = record.coverImage.ref.$link
  if (record.bskyPostRef) item.bskyPostRef = record.bskyPostRef

  await ddb.send(new PutCommand({ TableName: TABLE, Item: item }))
  // No fan-out: subscribers receive posts via fan-in when their subscriptions are processed in pass 2.
}

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,
      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 upsertSubscription(
  readerDid: string,
  rkey: string,
  record: { publication: string; createdAt?: string },
): Promise<void> {
  const pubAtUri = record.publication
  if (!pubAtUri?.startsWith('at://')) return
  const authorDid = pubAtUri.slice('at://'.length).split('/')[0]
  const subscribedAt = record.createdAt ?? new Date().toISOString()

  await ddb.send(new PutCommand({
    TableName: TABLE,
    Item: {
      PK: `u#${readerDid}#subscriptions`,
      SK: `sub#${rkey}`,
      entityType: 'subscription',
      publicationAtUri: pubAtUri,
      authorDid,
      subscribedAt,
    },
  }))

  // Conditionally write subscriber item; only increment count if the item is new.
  try {
    await ddb.send(new PutCommand({
      TableName: TABLE,
      Item: {
        PK: `u#${authorDid}#subscribers`,
        SK: `u#${readerDid}`,
        entityType: 'subscriber',
        subscribedAt,
      },
      ConditionExpression: 'attribute_not_exists(PK)',
    }))
    await ddb.send(new UpdateCommand({
      TableName: TABLE,
      Key: { PK: `u#${authorDid}`, SK: 'count' },
      UpdateExpression: 'ADD subscribers :inc',
      ExpressionAttributeValues: { ':inc': 1 },
    }))
  } catch (err) {
    if ((err as { name?: string }).name !== 'ConditionalCheckFailedException') throw err
  }

  await fanInExistingPosts(readerDid, authorDid, pubAtUri)
}

// ── PDS record fetching ───────────────────────────────────────────────────────

async function fetchAllRecords<T>(
  pdsUrl: string,
  did: string,
  collection: string,
): Promise<Array<{ uri: string; value: T }>> {
  const all: Array<{ uri: string; value: T }> = []
  let cursor: string | undefined

  do {
    const params = new URLSearchParams({ repo: did, collection, limit: '100' })
    if (cursor) params.set('cursor', cursor)
    const res = await fetch(`${pdsUrl}/xrpc/com.atproto.repo.listRecords?${params}`)
    if (!res.ok) throw new Error(`listRecords ${collection} for ${did}: ${res.status}`)
    const data = (await res.json()) as { records: Array<{ uri: string; value: T }>; cursor?: string }
    all.push(...data.records)
    cursor = data.cursor
  } while (cursor)

  return all
}

// ── Concurrency helper ────────────────────────────────────────────────────────

async function runConcurrent<T>(
  items: T[],
  limit: number,
  fn: (item: T) => Promise<void>,
): Promise<void> {
  const queue = [...items]
  const workers = Array.from({ length: Math.min(limit, items.length) }, async () => {
    while (queue.length > 0) {
      const item = queue.shift()!
      await fn(item)
    }
  })
  await Promise.all(workers)
}

// ── Per-DID backfill ──────────────────────────────────────────────────────────

type PubRecord = { url: string; name: string; description?: string; preferences?: { showInDiscover?: boolean } }
type DocRecord = {
  site: string; path?: string; title: string; description?: string; publishedAt: string; updatedAt?: string
  content?: { $type?: string; content?: string }
  coverImage?: { ref: { $link: string } }
  bskyPostRef?: { uri: string; cid: string }
}
type SubRecord = { publication: string; createdAt?: string }

async function backfillPublicationsAndDocuments(did: string): Promise<void> {
  let pdsUrl: string
  let resolvedHandle: string
  try {
    ;[pdsUrl, resolvedHandle] = await Promise.all([resolvePdsUrl(did), resolveHandle(did)])
  } catch (err) {
    console.error(`  skip ${did}: ${(err as Error).message}`)
    return
  }

  const [pubs, docs] = await Promise.all([
    fetchAllRecords<PubRecord>(pdsUrl, did, 'site.standard.publication').catch((err) => {
      console.error(`  publications fetch failed for ${did}: ${(err as Error).message}`)
      return []
    }),
    fetchAllRecords<DocRecord>(pdsUrl, did, 'site.standard.document').catch((err) => {
      console.error(`  documents fetch failed for ${did}: ${(err as Error).message}`)
      return []
    }),
  ])

  for (const { uri, value } of pubs) {
    const rkey = uri.split('/').pop()!
    try {
      await upsertPublication(did, rkey, value, resolvedHandle)
      console.log(`  pub  ${did}  ${rkey}  "${value.name}"`)
    } catch (err) {
      console.error(`  pub  ${did}  ${rkey}  ERROR: ${(err as Error).message}`)
    }
  }

  for (const { uri, value } of docs) {
    if (!value.site.startsWith('at://') && !ACCEPT_LOOSE_DOCUMENTS) {
      console.log(`  skip doc  ${did}  ${uri.split('/').pop()}  (loose document)`)
      continue
    }
    const rkey = uri.split('/').pop()!
    try {
      await upsertDocument(did, rkey, value, resolvedHandle)
      console.log(`  doc  ${did}  ${rkey}  "${value.title}"`)
    } catch (err) {
      console.error(`  doc  ${did}  ${rkey}  ERROR: ${(err as Error).message}`)
    }
  }
}

async function backfillSubscriptions(did: string): Promise<void> {
  let pdsUrl: string
  try {
    pdsUrl = await resolvePdsUrl(did)
  } catch (err) {
    console.error(`  skip ${did}: ${(err as Error).message}`)
    return
  }

  const subs = await fetchAllRecords<SubRecord>(
    pdsUrl, did, 'site.standard.graph.subscription',
  ).catch((err) => {
    console.error(`  subscriptions fetch failed for ${did}: ${(err as Error).message}`)
    return []
  })

  for (const { uri, value } of subs) {
    const rkey = uri.split('/').pop()!
    try {
      await upsertSubscription(did, rkey, value)
      console.log(`  sub  ${did}  ${rkey}  → ${value.publication}`)
    } catch (err) {
      console.error(`  sub  ${did}  ${rkey}  ERROR: ${(err as Error).message}`)
    }
  }
}

// ── Backfill core ─────────────────────────────────────────────────────────────

async function runBackfill(dids: string[]): Promise<void> {
  if (dids.length === 0) {
    console.log('No DIDs to process.')
    return
  }

  console.log(`Backfilling ${dids.length} DID(s)  concurrency=${CONCURRENCY}`)

  console.log('\n=== Pass 1: Publications & Documents ===')
  await runConcurrent(dids, CONCURRENCY, backfillPublicationsAndDocuments)

  console.log('\n=== Pass 2: Subscriptions ===')
  await runConcurrent(dids, CONCURRENCY, backfillSubscriptions)

  console.log('\nDone.')
}

// ── Lambda handler ────────────────────────────────────────────────────────────

type S3Event = {
  Records: Array<{ s3: { bucket: { name: string }; object: { key: string } } }>
}

export const handler = async (event: S3Event): Promise<void> => {
  const record = event.Records[0]
  const bucket = record.s3.bucket.name
  const key = decodeURIComponent(record.s3.object.key.replace(/\+/g, ' '))

  console.log(`Reading DIDs from s3://${bucket}/${key}`)
  const response = await s3.send(new GetObjectCommand({ Bucket: bucket, Key: key }))
  const body = await response.Body!.transformToString()

  const dids = body
    .split('\n')
    .map((l) => l.trim())
    .filter((l) => l && !l.startsWith('#'))

  await runBackfill(dids)
}

// ── CLI entry point ───────────────────────────────────────────────────────────

async function main(): Promise<void> {
  const didsFile = process.argv[2]
  if (!didsFile) {
    console.error('Usage: node dist/index.js <dids-file>')
    process.exit(1)
  }

  const dids = readFileSync(didsFile, 'utf8')
    .split('\n')
    .map((l) => l.trim())
    .filter((l) => l && !l.startsWith('#'))

  await runBackfill(dids)
}

if (require.main === module) {
  main().catch((err) => {
    console.error('Fatal:', err)
    process.exit(1)
  })
}
