feat: event log, pending operations, and MCP staging (#135)
* feat: event log, pending operations, and MCP staging - Event log system: persist bus events to event_log table for external automation platforms. Batch insert for transaction.synced. Daily cleanup cron at 02:00 UTC. - Pending operations: MCP write tools (categorize, create customer, create invoice) now stage to pending_operations instead of executing directly. Users review and commit/reject from /pending in the web UI. - Granskning page: card-based review UI with expandable previews, commit/reject dialogs. Only shown in nav when pending ops exist. - Commit route re-executes using core lib functions (no extension imports). Guards against stale state (double-commit, deleted entities). Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com> * feat: stage new MCP write tools after main merge Add staging for 4 new write tools from #133: - mark_invoice_paid, send_invoice, mark_invoice_sent, match_transaction_invoice - Expand pending_operations CHECK constraint - Add commit executors with full execution logic - Add UI labels and generic preview component - Remove confirm parameter from categorize (single-call staging) - Fix UUID in pending op title (fetch transaction description) - Hide Granskning nav when no pending ops Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com> * fix: address PR review feedback - Fix TS build error: use `select('*, customer:customers(*)')` for match_transaction_invoice to avoid array type inference - Add status guard to commitSendInvoice (prevents duplicate sends) - Replace auth.admin.getUserById with user email from session auth - Restore optimistic lock check in commitMatchTransactionInvoice - Fix tool description typo: expense_software → expense_office Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
Jakob Wennberg
co-authored by
Claude Opus 4.6
Claude Opus 4.6
parent
4d2152946a
commit
f3ae3cd361
@@ -536,3 +536,25 @@ export const PaginationQuerySchema = z.object({
|
||||
limit: z.coerce.number().int().min(1).max(100).default(50),
|
||||
offset: z.coerce.number().int().nonnegative().default(0),
|
||||
})
|
||||
|
||||
// ============================================================
|
||||
// Event log schemas
|
||||
// ============================================================
|
||||
|
||||
export const EventsQuerySchema = z.object({
|
||||
after: z.coerce.number().int().nonnegative().optional(),
|
||||
types: z.string()
|
||||
.transform(s => s.split(',').map(t => t.trim()).filter(Boolean))
|
||||
.optional(),
|
||||
limit: z.coerce.number().int().min(1).max(100).default(50),
|
||||
})
|
||||
|
||||
// ============================================================
|
||||
// Pending operations schemas
|
||||
// ============================================================
|
||||
|
||||
export const PendingOperationsQuerySchema = z.object({
|
||||
status: z.enum(['pending', 'committed', 'rejected']).default('pending'),
|
||||
limit: z.coerce.number().int().min(1).max(100).default(50),
|
||||
offset: z.coerce.number().int().nonnegative().default(0),
|
||||
})
|
||||
|
||||
@@ -0,0 +1,154 @@
|
||||
import { describe, it, expect, vi, beforeEach } from 'vitest'
|
||||
import { eventBus } from '@/lib/events/bus'
|
||||
import { makeTransaction, makeInvoice, makeCustomer, makeFiscalPeriod } from '@/tests/helpers'
|
||||
|
||||
// Mock the service client
|
||||
const mockInsert = vi.fn().mockResolvedValue({ error: null })
|
||||
vi.mock('@/lib/auth/api-keys', () => ({
|
||||
createServiceClientNoCookies: () => ({
|
||||
from: () => ({
|
||||
insert: mockInsert,
|
||||
}),
|
||||
}),
|
||||
}))
|
||||
|
||||
// Import after mocks
|
||||
import { registerEventLogHandler } from '../event-log-handler'
|
||||
|
||||
describe('event-log-handler', () => {
|
||||
let unsubscribers: (() => void)[]
|
||||
|
||||
beforeEach(() => {
|
||||
vi.clearAllMocks()
|
||||
eventBus.clear()
|
||||
mockInsert.mockResolvedValue({ error: null })
|
||||
unsubscribers = registerEventLogHandler()
|
||||
})
|
||||
|
||||
afterEach(() => {
|
||||
unsubscribers.forEach(unsub => unsub())
|
||||
})
|
||||
|
||||
it('persists invoice.created event with correct entity_id', async () => {
|
||||
const invoice = makeInvoice({ id: 'inv-123' })
|
||||
|
||||
await eventBus.emit({
|
||||
type: 'invoice.created',
|
||||
payload: { invoice, userId: 'user-1' },
|
||||
})
|
||||
|
||||
expect(mockInsert).toHaveBeenCalledTimes(1)
|
||||
expect(mockInsert).toHaveBeenCalledWith(
|
||||
expect.objectContaining({
|
||||
user_id: 'user-1',
|
||||
event_type: 'invoice.created',
|
||||
entity_id: 'inv-123',
|
||||
})
|
||||
)
|
||||
|
||||
// Data should NOT contain userId (it's in its own column)
|
||||
const insertedData = mockInsert.mock.calls[0][0].data
|
||||
expect(insertedData).not.toHaveProperty('userId')
|
||||
expect(insertedData).toHaveProperty('invoice')
|
||||
})
|
||||
|
||||
it('persists customer.created event', async () => {
|
||||
const customer = makeCustomer({ id: 'cust-456' })
|
||||
|
||||
await eventBus.emit({
|
||||
type: 'customer.created',
|
||||
payload: { customer, userId: 'user-1' },
|
||||
})
|
||||
|
||||
expect(mockInsert).toHaveBeenCalledTimes(1)
|
||||
expect(mockInsert).toHaveBeenCalledWith(
|
||||
expect.objectContaining({
|
||||
user_id: 'user-1',
|
||||
event_type: 'customer.created',
|
||||
entity_id: 'cust-456',
|
||||
})
|
||||
)
|
||||
})
|
||||
|
||||
it('batch-inserts transaction.synced as individual rows', async () => {
|
||||
const tx1 = makeTransaction({ id: 'tx-1' })
|
||||
const tx2 = makeTransaction({ id: 'tx-2' })
|
||||
const tx3 = makeTransaction({ id: 'tx-3' })
|
||||
|
||||
await eventBus.emit({
|
||||
type: 'transaction.synced',
|
||||
payload: { transactions: [tx1, tx2, tx3], userId: 'user-1' },
|
||||
})
|
||||
|
||||
expect(mockInsert).toHaveBeenCalledTimes(1)
|
||||
const rows = mockInsert.mock.calls[0][0]
|
||||
expect(rows).toHaveLength(3)
|
||||
expect(rows[0]).toMatchObject({ event_type: 'transaction.synced', entity_id: 'tx-1' })
|
||||
expect(rows[1]).toMatchObject({ event_type: 'transaction.synced', entity_id: 'tx-2' })
|
||||
expect(rows[2]).toMatchObject({ event_type: 'transaction.synced', entity_id: 'tx-3' })
|
||||
})
|
||||
|
||||
it('does NOT persist journal_entry.drafted (excluded noise event)', async () => {
|
||||
await eventBus.emit({
|
||||
type: 'journal_entry.drafted',
|
||||
payload: { entry: {} as unknown, userId: 'user-1' },
|
||||
})
|
||||
|
||||
expect(mockInsert).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it('does NOT persist receipt.extracted (excluded noise event)', async () => {
|
||||
await eventBus.emit({
|
||||
type: 'receipt.extracted',
|
||||
payload: { receipt: {} as unknown, documentId: null, confidence: 0.9, userId: 'user-1' },
|
||||
})
|
||||
|
||||
expect(mockInsert).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it('does NOT persist supplier_invoice.received (excluded noise event)', async () => {
|
||||
await eventBus.emit({
|
||||
type: 'supplier_invoice.received',
|
||||
payload: { inboxItem: {} as unknown, userId: 'user-1' },
|
||||
})
|
||||
|
||||
expect(mockInsert).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it('does NOT persist supplier_invoice.extracted (excluded noise event)', async () => {
|
||||
await eventBus.emit({
|
||||
type: 'supplier_invoice.extracted',
|
||||
payload: { inboxItem: {} as unknown, confidence: 0.9, userId: 'user-1' },
|
||||
})
|
||||
|
||||
expect(mockInsert).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it('does not throw when persistence fails', async () => {
|
||||
mockInsert.mockResolvedValue({ error: { message: 'DB down' } })
|
||||
|
||||
// Should not throw
|
||||
await eventBus.emit({
|
||||
type: 'customer.created',
|
||||
payload: { customer: makeCustomer(), userId: 'user-1' },
|
||||
})
|
||||
|
||||
expect(mockInsert).toHaveBeenCalledTimes(1)
|
||||
})
|
||||
|
||||
it('persists period.locked with period entity_id', async () => {
|
||||
const period = makeFiscalPeriod({ id: 'period-1' })
|
||||
|
||||
await eventBus.emit({
|
||||
type: 'period.locked',
|
||||
payload: { period, userId: 'user-1' },
|
||||
})
|
||||
|
||||
expect(mockInsert).toHaveBeenCalledWith(
|
||||
expect.objectContaining({
|
||||
event_type: 'period.locked',
|
||||
entity_id: 'period-1',
|
||||
})
|
||||
)
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,145 @@
|
||||
import { eventBus } from '@/lib/events/bus'
|
||||
import type { CoreEventType } from '@/lib/events/types'
|
||||
import { createServiceClientNoCookies } from '@/lib/auth/api-keys'
|
||||
import { createLogger } from '@/lib/logger'
|
||||
|
||||
const log = createLogger('event-log')
|
||||
|
||||
/**
|
||||
* Event types persisted to the event_log table for external automation platforms.
|
||||
* Excludes noise events that are always followed by an actionable event.
|
||||
*/
|
||||
const PERSISTED_EVENT_TYPES: CoreEventType[] = [
|
||||
'journal_entry.committed',
|
||||
'journal_entry.corrected',
|
||||
'document.uploaded',
|
||||
'invoice.created',
|
||||
'invoice.sent',
|
||||
'credit_note.created',
|
||||
'transaction.synced',
|
||||
'transaction.categorized',
|
||||
'transaction.reconciled',
|
||||
'period.locked',
|
||||
'period.year_closed',
|
||||
'customer.created',
|
||||
'receipt.matched',
|
||||
'receipt.confirmed',
|
||||
'supplier_invoice.registered',
|
||||
'supplier_invoice.approved',
|
||||
'supplier_invoice.paid',
|
||||
'supplier_invoice.credited',
|
||||
'invoice.match_confirmed',
|
||||
'supplier_invoice.match_confirmed',
|
||||
'supplier_invoice.confirmed',
|
||||
]
|
||||
|
||||
// Excluded (with reasoning):
|
||||
// - journal_entry.drafted: always followed by .committed
|
||||
// - receipt.extracted: intermediate AI step; .matched/.confirmed are actionable
|
||||
// - supplier_invoice.received: inbox receipt; .confirmed is actionable
|
||||
// - supplier_invoice.extracted: intermediate AI step
|
||||
|
||||
/**
|
||||
* Extract the primary entity ID from an event payload.
|
||||
*/
|
||||
function extractEntityId(payload: Record<string, unknown>): string | null {
|
||||
// Try common entity shapes in priority order
|
||||
const entityKeys = [
|
||||
'entry', 'invoice', 'transaction', 'customer', 'receipt',
|
||||
'supplierInvoice', 'creditNote', 'period', 'document', 'inboxItem',
|
||||
] as const
|
||||
|
||||
for (const key of entityKeys) {
|
||||
const entity = payload[key]
|
||||
if (entity && typeof entity === 'object' && 'id' in entity) {
|
||||
const id = (entity as Record<string, unknown>).id
|
||||
if (typeof id === 'string') return id
|
||||
}
|
||||
}
|
||||
|
||||
// For journal_entry.corrected: use the corrected entry's ID
|
||||
if ('corrected' in payload) {
|
||||
const corrected = payload.corrected
|
||||
if (corrected && typeof corrected === 'object' && 'id' in corrected) {
|
||||
const id = (corrected as Record<string, unknown>).id
|
||||
if (typeof id === 'string') return id
|
||||
}
|
||||
}
|
||||
|
||||
return null
|
||||
}
|
||||
|
||||
/**
|
||||
* Strip userId from payload (stored in its own column) and return clean data.
|
||||
*/
|
||||
function stripUserId(payload: Record<string, unknown>): Record<string, unknown> {
|
||||
const { userId: _userId, ...data } = payload
|
||||
return data
|
||||
}
|
||||
|
||||
/**
|
||||
* Persist a single event to the event_log table.
|
||||
*/
|
||||
async function persistEvent(
|
||||
eventType: string,
|
||||
userId: string,
|
||||
entityId: string | null,
|
||||
data: Record<string, unknown>
|
||||
): Promise<void> {
|
||||
const supabase = createServiceClientNoCookies()
|
||||
|
||||
const { error } = await supabase
|
||||
.from('event_log')
|
||||
.insert({
|
||||
user_id: userId,
|
||||
event_type: eventType,
|
||||
entity_id: entityId,
|
||||
data,
|
||||
})
|
||||
|
||||
if (error) {
|
||||
log.error(`Failed to persist event ${eventType}:`, error.message)
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Register event log handlers on the event bus.
|
||||
* Persists events to the event_log table for external automation platforms.
|
||||
* Returns an array of unsubscribe functions.
|
||||
*/
|
||||
export function registerEventLogHandler(): (() => void)[] {
|
||||
return PERSISTED_EVENT_TYPES.map((eventType) =>
|
||||
eventBus.on(eventType, async (payload) => {
|
||||
try {
|
||||
const rawPayload = payload as Record<string, unknown>
|
||||
const userId = rawPayload.userId as string
|
||||
|
||||
// transaction.synced carries an array — batch insert
|
||||
if (eventType === 'transaction.synced' && Array.isArray(rawPayload.transactions)) {
|
||||
const transactions = rawPayload.transactions as Array<Record<string, unknown>>
|
||||
if (transactions.length === 0) return
|
||||
|
||||
const rows = transactions.map(tx => ({
|
||||
user_id: userId,
|
||||
event_type: eventType,
|
||||
entity_id: typeof tx.id === 'string' ? tx.id : null,
|
||||
data: { transaction: tx },
|
||||
}))
|
||||
|
||||
const supabase = createServiceClientNoCookies()
|
||||
const { error } = await supabase.from('event_log').insert(rows)
|
||||
if (error) {
|
||||
log.error(`Failed to persist batch transaction.synced:`, error.message)
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
const entityId = extractEntityId(rawPayload)
|
||||
const data = stripUserId(rawPayload)
|
||||
await persistEvent(eventType, userId, entityId, data)
|
||||
} catch (err) {
|
||||
log.error(`Event log handler error for ${eventType}:`, err)
|
||||
}
|
||||
})
|
||||
)
|
||||
}
|
||||
@@ -2,6 +2,7 @@ import { loadExtensions } from '@/lib/extensions/loader'
|
||||
import { setContextFactory } from '@/lib/extensions/registry'
|
||||
import { createExtensionContext } from '@/lib/extensions/context-factory'
|
||||
import { registerSupplierInvoiceHandler } from '@/lib/bookkeeping/handlers/supplier-invoice-handler'
|
||||
import { registerEventLogHandler } from '@/lib/events/handlers/event-log-handler'
|
||||
import { createLogger } from '@/lib/logger'
|
||||
|
||||
const log = createLogger('init')
|
||||
@@ -74,6 +75,7 @@ export function ensureInitialized(): void {
|
||||
validateEnvironment()
|
||||
setContextFactory(createExtensionContext)
|
||||
registerSupplierInvoiceHandler()
|
||||
registerEventLogHandler()
|
||||
loadExtensions()
|
||||
|
||||
initialized = true
|
||||
|
||||
Reference in New Issue
Block a user