01
import { Hono } from 'hono'
02
import { dailyUsageRollup, cleanupExpiredSessions } from './cron/jobs'
03

04
type Bindings = {
05
  DB: D1Database
06
  USAGE_KV: KVNamespace
07
}
08

09
const app = new Hono<{ Bindings: Bindings }>()
10

11
// 正常的 HTTP 路由
12
app.get('/', (c) => c.text('OK'))
13

14
// 包装默认导出，把 fetch 和 scheduled 放在一起
15
export default {
16
  fetch: app.fetch,
17
  // scheduled 不经过 Hono，所以没有 c（Context）
18
  // env 和 ctx 从函数参数直接拿
19
  async scheduled(
20
    controller: ScheduledController,  // Workers 内置类型
21
    env: Bindings,
22
    ctx: ExecutionContext              // Workers 内置类型
23
  ) {
24
    // 根据 cron 表达式分发到不同任务
25
    switch (controller.cron) {
26
      case '7 * * * *':
27
        ctx.waitUntil(dailyUsageRollup(env))
28
        break
29
      case '30 3 * * *':
30
        ctx.waitUntil(cleanupExpiredSessions(env))
31
        break
32
    }
33
  },
34
}

01
import { drizzle } from 'drizzle-orm/d1'
02
import { usageDaily } from '../db/schema'
03

04
export async function dailyUsageRollup(env: {
05
  DB: D1Database
06
  USAGE_KV: KVNamespace
07
}) {
08
  const db = drizzle(env.DB)
09
  const today = new Date().toISOString().slice(0, 10)
10

11
  // 列出所有用户的当天用量（KV 里按前缀扫）
12
  const list = await env.USAGE_KV.list({ prefix: `usage:` })
13

14
  for (const key of list.keys) {
15
    // key 格式：usage:<apiKeyId>:<date>
16
    const parts = key.name.split(':')
17
    if (parts.length !== 3 || parts[2] !== today) continue
18

19
    const apiKeyId = parts[1]
20
    const record = await env.USAGE_KV.get<{
21
      totalRequests: number
22
      totalPromptTokens: number
23
      totalCompletionTokens: number
24
    }>(key.name, 'json')
25

26
    if (!record) continue
27

28
    // 写进 D1 的日表
29
    await db.insert(usageDaily).values({
30
      apiKeyId,
31
      date: today,
32
      requests: record.totalRequests,
33
      promptTokens: record.totalPromptTokens,
34
      completionTokens: record.totalCompletionTokens,
35
    }).onConflictDoUpdate({
36
      target: [usageDaily.apiKeyId, usageDaily.date],
37
      set: {
38
        requests: record.totalRequests,
39
        promptTokens: record.totalPromptTokens,
40
        completionTokens: record.totalCompletionTokens,
41
      },
42
    })
43
  }
44
}

01
import { Hono } from 'hono'
02

03
type Bindings = {
04
  INGEST_QUEUE: Queue<IngestJob>
05
}
06

07
interface IngestJob {
08
  docId: string
09
  text: string
10
  tenantId: string
11
}
12

13
const upload = new Hono<{ Bindings: Bindings }>()
14

15
upload.post('/documents', async (c) => {
16
  const { docs, tenantId } = await c.req.json<{
17
    docs: Array<{ id: string; text: string }>
18
    tenantId: string
19
  }>()
20

21
  // 不在这里做 embedding（可能要几秒到几十秒）
22
  // 只是把「要做的事」塞进队列，立即返回
23
  for (const doc of docs) {
24
    await c.env.INGEST_QUEUE.send({
25
      docId: doc.id,
26
      text: doc.text,
27
      tenantId,
28
    })
29
  }
30

31
  return c.json({
32
    accepted: docs.length,
33
    message: '已接收，正在后台处理',
34
  }, 202)
35
})
36

37
export default upload

01
interface IngestJob {
02
  docId: string
03
  text: string
04
  tenantId: string
05
}
06

07
type Env = {
08
  AI: Ai
09
  DOCS_INDEX: Vectorize
10
}
11

12
export async function ingestConsumer(
13
  batch: MessageBatch<IngestJob>,
14
  env: Env
15
) {
16
  // 1. 批量生成 embedding（一次调用比一条一条调便宜很多）
17
  const texts = batch.messages.map((m) => m.body.text)
18
  const { data: embeddings } = await env.AI.run(
19
    '@cf/baai/bge-base-en-v1.5',
20
    { text: texts }
21
  )
22

23
  // 2. 组装写入向量数据库
24
  const vectors = batch.messages.map((m, i) => ({
25
    id: m.body.docId,
26
    values: embeddings[i],
27
    namespace: m.body.tenantId,
28
    metadata: {
29
      text: m.body.text,
30
    },
31
  }))
32

33
  try {
34
    await env.DOCS_INDEX.upsert(vectors)
35
    // 批量 ack：告诉 Queue 这一批都成功了
36
    for (const msg of batch.messages) msg.ack()
37
  } catch (err) {
38
    // 整批重试。也可以逐条 retry() 做更精细的控制
39
    for (const msg of batch.messages) msg.retry({ delaySeconds: 30 })
40
  }
41
}

01
import { Hono } from 'hono'
02
import { ingestConsumer } from './consumers/ingest'
03
import upload from './routes/upload'
04

05
const app = new Hono()
06
app.route('/upload', upload)
07

08
export default {
09
  fetch: app.fetch,
10
  async queue(batch: MessageBatch, env: any, ctx: ExecutionContext) {
11
    // 通过 queue 名字分发到不同 consumer
12
    if (batch.queue === 'rag-ingest') {
13
      return ingestConsumer(batch, env)
14
    }
15
  },
16
}

01
export async function checkSubscriptions(env: {
02
  DB: D1Database
03
  CHECK_QUEUE: Queue<{ userId: number }>
04
}) {
05
  const db = drizzle(env.DB)
06

07
  // Cron 只做「分发」，每 1000 条用户推一批到队列
08
  let lastId = 0
09
  while (true) {
10
    const batch = await db
11
      .select({ id: users.id })
12
      .from(users)
13
      .where(gt(users.id, lastId))
14
      .limit(1000)
15
      .all()
16

17
    if (batch.length === 0) break
18

19
    // 批量 send（一次 API 调用塞多条）
20
    await env.CHECK_QUEUE.sendBatch(
21
      batch.map((u) => ({ body: { userId: u.id } }))
22
    )
23
    lastId = batch[batch.length - 1].id
24
  }
25
}

1
const seen = await env.KV.get(`processed:${msg.body.docId}`)
2
if (seen) {
3
  msg.ack()
4
  continue
5
}
6
// ... 实际处理 ...
7
await env.KV.put(`processed:${msg.body.docId}`, '1', { expirationTtl: 86400 })
8
msg.ack()

1
await c.env.CHECK_PAYMENT_QUEUE.send(
2
  { orderId: '123' },
3
  { delaySeconds: 30 * 60 }
4
)