1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

// src/db/schema.ts

export const creditTransactionTable = sqliteTable("credit_transactions", {
  id: text("id").primaryKey(),
  userId: text("user_id").notNull().references(() => userTable.id),
  amount: integer("amount").notNull(),           // 原始金额
  remainingAmount: integer("remaining_amount").notNull(), // 剩余金额
  type: text("type", {
    enum: ["purchase", "consumption", "expiration", "refund"],
  }).notNull(),
  description: text("description"),
  expirationDate: integer("expiration_date", { mode: "timestamp" }),
  // 过期处理标记
  expirationDateProcessedAt: integer("expiration_date_processed_at",
    { mode: "timestamp" }),
  createdAt: integer("created_at", { mode: "timestamp" })
    .notNull().$defaultFn(() => new Date()),
});

export const userTable = sqliteTable("users", {
  // ...
  currentCredits: integer("current_credits").notNull().default(0),
  lastCreditRefreshAt: integer("last_credit_refresh_at",
    { mode: "timestamp" }),
});

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63

// src/utils/credits.ts

// 实际函数名：consumeUserCredits，参数为对象
export async function consumeUserCredits({
  userId,
  amount,
  description,
}: {
  userId: string;
  amount: number;
  description: string;
}) {
  const db = getDB();
  const currentTime = new Date();

  // 先处理过期积分
  await processExpiredCredits(userId, currentTime);

  // 查询可用积分桶，排序：先按过期时间（最快过期的先消费），再按创建时间
  const sourceTransactions = await db.query.creditTransactionTable.findMany({
    where: and(
      eq(creditTransactionTable.userId, userId),
      inArray(creditTransactionTable.type, [
        CREDIT_TRANSACTION_TYPE.PURCHASE,
        CREDIT_TRANSACTION_TYPE.MONTHLY_REFRESH,
      ]),
      gt(creditTransactionTable.remainingAmount, 0),
      isNull(creditTransactionTable.expirationDateProcessedAt),
      or(
        isNull(creditTransactionTable.expirationDate),
        gte(creditTransactionTable.expirationDate, currentTime),
      ),
    ),
    orderBy: [
      asc(creditTransactionTable.expirationDate), // 最快过期的先消费
      asc(creditTransactionTable.createdAt),
    ],
    columns: { id: true, remainingAmount: true },
  });

  const availableFromBuckets = sourceTransactions.reduce(
    (total, t) => total + t.remainingAmount, 0
  );

  if (availableFromBuckets < amount) {
    throw new InsufficientCreditsError({
      requiredCredits: amount,
      availableCredits: availableFromBuckets,
    });
  }

  // 从桶中扣减（FIFO，按过期时间优先）
  // ...（逐桶扣减逻辑）

  // 更新用户总积分
  await db
    .update(userTable)
    .set({ currentCredits: sql`${userTable.currentCredits} - ${amount}` })
    .where(and(
      eq(userTable.id, userId),
      gte(userTable.currentCredits, amount),
    ));
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42

async function processExpiredCredits(userId: string, currentTime: Date) {
  const db = getDB();

  // 查找所有已过期但未处理的积分
  const expiredTransactions = await db.query.creditTransactionTable.findMany({
    where: and(
      eq(creditTransactionTable.userId, userId),
      lt(creditTransactionTable.expirationDate, currentTime),
      isNull(creditTransactionTable.expirationDateProcessedAt),
      gt(creditTransactionTable.remainingAmount, 0),
    ),
    orderBy: [asc(creditTransactionTable.createdAt)],
  });

  for (const transaction of expiredTransactions) {
    // 原子更新：只有 remainingAmount 没变才更新
    const updateResult = await db
      .update(creditTransactionTable)
      .set({
        expirationDateProcessedAt: currentTime,
        remainingAmount: 0,
      })
      .where(and(
        eq(creditTransactionTable.id, transaction.id),
        isNull(creditTransactionTable.expirationDateProcessedAt),
        // 乐观锁：只在 remainingAmount 没变时才更新
        eq(creditTransactionTable.remainingAmount, transaction.remainingAmount)
      ))
      .returning({ id: creditTransactionTable.id });

    // 如果没更新成功，说明被其他请求处理了
    if (!updateResult || updateResult.length === 0) {
      continue;
    }

    // 从用户总积分中扣除
    await db
      .update(userTable)
      .set({ currentCredits: sql`current_credits - ${transaction.remainingAmount}` })
      .where(eq(userTable.id, userId));
  }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

// src/utils/auth.ts

async function validateSessionToken(token: string, userId: string) {
  const session = await getKVSession(sessionId, userId);

  // 同步积分
  const currentCredits = await syncCreditsIfNeeded(session);

  if (
    session?.user?.currentCredits &&
    currentCredits !== session.user.currentCredits
  ) {
    session.user.currentCredits = currentCredits;
  }

  return session;
}