nanakura/moeka-project
1
0
Fork 0
Code Issues Pull Requests 1 Actions 2 Packages Projects Releases Wiki Activity

refactor(server): squash flux-audit-log into flux-ledger

Browse Source
This commit is contained in:
RainbowBird
2026-03-28 02:25:44 +08:00
committed by RainbowBird
parent 0a2fefc304
commit 11c456c0a7
12 changed files with 2304 additions and 67 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/server/drizzle/0006_overconfident_susan_delgado.sql
+2
View File
@@ -0,0 +1,2 @@
DROP TABLE "flux_audit_log" CASCADE;--> statement-breakpoint
ALTER TABLE "flux_ledger" ADD COLUMN "metadata" jsonb;
apps/server/drizzle/meta/0006_snapshot.json
+2270
View File
File diff suppressed because it is too large Load Diff
apps/server/drizzle/meta/_journal.json
+7
View File
@@ -43,6 +43,13 @@
"when": 1774582846222,
"tag": "0005_tough_living_tribunal",
"breakpoints": true
},
{
"idx": 6,
"version": "7",
"when": 1774584037626,
"tag": "0006_overconfident_susan_delgado",
"breakpoints": true
}
]
}
apps/server/src/schemas/flux-audit-log.ts
-14
View File
@@ -1,14 +0,0 @@
import { integer, jsonb, pgTable, text, timestamp } from 'drizzle-orm/pg-core'
import { nanoid } from '../utils/id'
import { user } from './accounts'
export const fluxAuditLog = pgTable('flux_audit_log', {
id: text('id').primaryKey().$defaultFn(() => nanoid()),
userId: text('user_id').notNull().references(() => user.id, { onDelete: 'cascade' }),
type: text('type').notNull(), // 'consumption' | 'addition' | 'initial'
amount: integer('amount').notNull(), // positive = gain, negative = spend
description: text('description').notNull(), // model name, "Stripe payment", "Initial grant", etc.
metadata: jsonb('metadata'), // { promptTokens, completionTokens, stripeSessionId, ... }
createdAt: timestamp('created_at').defaultNow().notNull(),
})
apps/server/src/schemas/flux-ledger.ts
+2 -1
View File
@@ -1,5 +1,5 @@
import { sql } from 'drizzle-orm'
import { index, integer, pgTable, text, timestamp, uniqueIndex } from 'drizzle-orm/pg-core'
import { index, integer, jsonb, pgTable, text, timestamp, uniqueIndex } from 'drizzle-orm/pg-core'
import { nanoid } from '../utils/id'
import { user } from './accounts'
@@ -13,6 +13,7 @@ export const fluxLedger = pgTable('flux_ledger', {
balanceAfter: integer('balance_after').notNull(),
requestId: text('request_id'), // nullable; used for idempotency on debit/credit
description: text('description').notNull(),
metadata: jsonb('metadata'), // { promptTokens, completionTokens, stripeSessionId, ... }
createdAt: timestamp('created_at').defaultNow().notNull(),
}, table => [
index('flux_ledger_user_id_idx').on(table.userId),
apps/server/src/schemas/index.ts
-1
View File
@@ -2,7 +2,6 @@ export * from './accounts'
export * from './characters'
export * from './chats'
export * from './flux'
export * from './flux-audit-log'
export * from './flux-ledger'
export * from './llm-request-log'
export * from './providers'
apps/server/src/services/billing/billing-consumer-handler.ts
-8
View File
@@ -3,7 +3,6 @@ import type { BillingStreamMessage } from './billing-mq'
import { useLogger } from '@guiiai/logg'
import * as fluxAuditSchema from '../../schemas/flux-audit-log'
import * as fluxLedgerSchema from '../../schemas/flux-ledger'
import * as llmRequestLogSchema from '../../schemas/llm-request-log'
@@ -30,13 +29,6 @@ export function createBillingConsumerHandler(db: Database) {
description: event.payload.source ?? 'LLM request',
})
await db.insert(fluxAuditSchema.fluxAuditLog).values({
userId: event.userId,
type: 'consumption',
amount: -event.payload.amount,
description: event.payload.source ?? 'LLM request',
})
logger.withFields({
eventId: event.eventId,
userId: event.userId,
apps/server/src/services/billing/billing-service.ts
-25
View File
@@ -14,7 +14,6 @@ import { nanoid } from '../../utils/id'
import { fluxRedisKey } from '../flux'
import * as fluxSchema from '../../schemas/flux'
import * as fluxAuditSchema from '../../schemas/flux-audit-log'
import * as fluxLedgerSchema from '../../schemas/flux-ledger'
import * as stripeSchema from '../../schemas/stripe'
@@ -164,14 +163,6 @@ export function createBillingService(
balanceAfter,
requestId: input.requestId,
description: input.description,
})
// Audit log
await tx.insert(fluxAuditSchema.fluxAuditLog).values({
userId: input.userId,
type: 'addition',
amount: input.amount,
description: input.description,
metadata: input.auditMetadata,
})
@@ -258,14 +249,6 @@ export function createBillingService(
balanceAfter,
requestId: input.stripeEventId,
description,
})
// Audit log
await tx.insert(fluxAuditSchema.fluxAuditLog).values({
userId: input.userId,
type: 'addition',
amount: input.fluxAmount,
description,
metadata: {
stripeEventId: input.stripeEventId,
stripeSessionId: input.stripeSessionId,
@@ -374,14 +357,6 @@ export function createBillingService(
balanceAfter,
requestId: input.stripeEventId,
description,
})
// Audit log
await tx.insert(fluxAuditSchema.fluxAuditLog).values({
userId: input.userId,
type: 'addition',
amount: input.fluxAmount,
description,
metadata: {
stripeEventId: input.stripeEventId,
stripeInvoiceId: input.stripeInvoiceId,
apps/server/src/services/billing/tests/billing-service.test.ts
+6 -5
View File
@@ -62,7 +62,6 @@ describe('billingService', () => {
billingMq = createMockBillingMq()
billingService = createBillingService(db, redis, billingMq, createMockConfigKV())
await db.delete(schema.fluxAuditLog)
await db.delete(schema.fluxLedger)
await db.delete(schema.userFlux).where(eq(schema.userFlux.userId, 'user-billing-1'))
await db.delete(schema.stripeCheckoutSession).where(eq(schema.stripeCheckoutSession.stripeSessionId, 'sess-billing-1'))
@@ -103,10 +102,12 @@ describe('billingService', () => {
expect(ledgerRecords[0]?.balanceBefore).toBe(0)
expect(ledgerRecords[0]?.balanceAfter).toBe(50)
// Verify audit log
const auditRecords = await db.select().from(schema.fluxAuditLog).where(eq(schema.fluxAuditLog.userId, 'user-billing-1'))
expect(auditRecords).toHaveLength(1)
expect(auditRecords[0]?.amount).toBe(50)
// Verify metadata on ledger entry
expect(ledgerRecords[0]?.metadata).toMatchObject({
stripeEventId: 'stripe-evt-1',
stripeSessionId: 'sess-billing-1',
source: 'stripe.checkout.completed',
})
// Verify billing events published to stream
expect(billingMq.publish).toHaveBeenCalledTimes(2)
apps/server/src/services/flux-audit.ts
+10 -7
View File
@@ -3,14 +3,17 @@ import type { Database } from '../libs/db'
import { useLogger } from '@guiiai/logg'
import { desc, eq } from 'drizzle-orm'
import * as schema from '../schemas/flux-audit-log'
import * as schema from '../schemas/flux-ledger'
const logger = useLogger('flux-audit')
export interface AuditEntry {
userId: string
type: 'consumption' | 'addition' | 'initial'
type: 'credit' | 'debit' | 'initial'
amount: number
balanceBefore: number
balanceAfter: number
requestId?: string
description: string
metadata?: Record<string, unknown>
}
@@ -18,21 +21,21 @@ export interface AuditEntry {
export function createFluxAuditService(db: Database) {
return {
async log(entry: AuditEntry) {
await db.insert(schema.fluxAuditLog).values(entry)
await db.insert(schema.fluxLedger).values(entry)
logger.withFields({ userId: entry.userId, type: entry.type, amount: entry.amount }).log('Audit entry recorded')
},
async logBatch(entries: AuditEntry[]) {
if (entries.length === 0)
return
await db.insert(schema.fluxAuditLog).values(entries)
await db.insert(schema.fluxLedger).values(entries)
logger.withFields({ count: entries.length }).log('Audit batch recorded')
},
async getHistory(userId: string, limit: number, offset: number) {
const records = await db.query.fluxAuditLog.findMany({
where: eq(schema.fluxAuditLog.userId, userId),
orderBy: [desc(schema.fluxAuditLog.createdAt)],
const records = await db.query.fluxLedger.findMany({
where: eq(schema.fluxLedger.userId, userId),
orderBy: [desc(schema.fluxLedger.createdAt)],
limit: limit + 1, // fetch one extra to determine hasMore
offset,
})
apps/server/src/services/tests/flux-audit.test.ts
+7 -5
View File
@@ -19,25 +19,27 @@ describe('fluxAuditService', () => {
service = createFluxAuditService(db)
})
it('log should insert a single audit entry', async () => {
it('log should insert a single ledger entry', async () => {
await service.log({
userId: 'user-audit',
type: 'addition',
type: 'credit',
amount: 500,
balanceBefore: 0,
balanceAfter: 500,
description: 'Stripe payment',
metadata: { stripeSessionId: 'sess_123' },
})
const { records } = await service.getHistory('user-audit', 10, 0)
expect(records).toHaveLength(1)
expect(records[0].type).toBe('addition')
expect(records[0].type).toBe('credit')
expect(records[0].amount).toBe(500)
})
it('logBatch should insert multiple entries', async () => {
await service.logBatch([
{ userId: 'user-audit', type: 'consumption', amount: -10, description: 'gpt-4o' },
{ userId: 'user-audit', type: 'consumption', amount: -5, description: 'gpt-4o-mini' },
{ userId: 'user-audit', type: 'debit', amount: 10, balanceBefore: 500, balanceAfter: 490, description: 'gpt-4o' },
{ userId: 'user-audit', type: 'debit', amount: 5, balanceBefore: 490, balanceAfter: 485, description: 'gpt-4o-mini' },
])
const { records } = await service.getHistory('user-audit', 10, 0)
apps/server/src/services/tests/flux.test.ts
-1
View File
@@ -55,7 +55,6 @@ describe('fluxService (DB-backed)', () => {
// Clean up flux-related tables
await db.delete(schema.fluxLedger).where(eq(schema.fluxLedger.userId, testUser.id))
await db.delete(schema.fluxAuditLog).where(eq(schema.fluxAuditLog.userId, testUser.id))
await db.delete(schema.userFlux).where(eq(schema.userFlux.userId, testUser.id))
})