The Content Pipeline: From Notion to Published in Four Stages

import { Client } from '@notionhq/client'

class RateLimiter {
  private queue: Array<() => void> = []
  private processing = false
  private readonly interval: number

  constructor(requestsPerSecond: number) {
    this.interval = Math.ceil(1000 / requestsPerSecond)
  }

  async schedule<T>(fn: () => Promise<T>): Promise<T> {
    return new Promise((resolve, reject) => {
      this.queue.push(async () => {
        try {
          resolve(await fn())
        } catch (err) {
          reject(err)
        }
      })
      this.processQueue()
    })
  }

  private async processQueue() {
    if (this.processing) return
    this.processing = true
    while (this.queue.length > 0) {
      const next = this.queue.shift()!
      await next()
      await new Promise((r) => setTimeout(r, this.interval))
    }
    this.processing = false
  }
}

const limiter = new RateLimiter(3)
const notion = new Client({ auth: process.env.NOTION_API_KEY })

export async function queryDatabase(databaseId: string, filter?: any) {
  const results: any[] = []
  let cursor: string | undefined

  do {
    const response = await limiter.schedule(() =>
      notion.databases.query({
        database_id: databaseId,
        filter,
        start_cursor: cursor,
        page_size: 100,
      })
    )
    results.push(...response.results)
    cursor = response.has_more ? response.next_cursor! : undefined
  } while (cursor)

  return results
}

async function generateMetaTags(record: ContentRecord) {
  const prompt = `Generate SEO metadata for the following article.

Title: ${record.title}
Excerpt: ${record.excerpt}
Content (first 500 words): ${record.content.slice(0, 3).map(b => b.text).join(' ')}

Requirements:
- meta_title: max 60 characters, include primary keyword
- meta_description: max 155 characters, complete sentence
- og_title: can differ from meta_title, max 70 characters
- keywords: array of 5-8 relevant terms, no generic words

Return JSON only. No explanation.`

  const result = await llm.generate({
    model: 'claude-sonnet',
    prompt,
    temperature: 0.2,
    max_tokens: 300,
  })

  return JSON.parse(result.text)
}

interface PipelineResult {
  stage: 'connect' | 'source' | 'creative' | 'publish'
  recordId: string
  status: 'success' | 'failed' | 'skipped'
  error?: string
  duration: number
  timestamp: string
}

async function runPipeline(options: { recordIds?: string[] }) {
  const results: PipelineResult[] = []

  // Stage 1: Connect
  const rawPages = await connectStage(options.recordIds)

  // Stage 2: Source
  for (const page of rawPages) {
    try {
      const record = await sourceStage(page)
      if (record.status !== 'approved' && record.status !== 'published') {
        results.push({
          stage: 'source', recordId: record.id,
          status: 'skipped', duration: 0,
          timestamp: new Date().toISOString(),
        })
        continue
      }

      // Stage 3: Creative
      const enhanced = await creativeStage(record)

      // Stage 4: Publish
      await publishStage(enhanced)
      results.push({
        stage: 'publish', recordId: record.id,
        status: 'success', duration: Date.now(),
        timestamp: new Date().toISOString(),
      })
    } catch (err) {
      results.push({
        stage: 'source', recordId: page.id,
        status: 'failed', error: String(err),
        duration: 0, timestamp: new Date().toISOString(),
      })
    }
  }
  return results
}