From d8f707ef23f94f8336bc961e3c9b5caff4f491f4 Mon Sep 17 00:00:00 2001 From: Shotaro Nakamura Date: Tue, 14 Jul 2026 15:17:14 +0900 Subject: [PATCH 1/7] Add Mobile Suica history range sync --- apps/mnie-server/src/providers/registry.ts | 3 + docs/docs/providers/mobilesuica.md | 17 ++- packages/provider-mobile-suica/README.md | 4 +- packages/provider-mobile-suica/src/index.ts | 149 ++++++++++++++++++-- 4 files changed, 159 insertions(+), 14 deletions(-) diff --git a/apps/mnie-server/src/providers/registry.ts b/apps/mnie-server/src/providers/registry.ts index f648a3b..8dc0f90 100644 --- a/apps/mnie-server/src/providers/registry.ts +++ b/apps/mnie-server/src/providers/registry.ts @@ -776,6 +776,9 @@ export class ProviderRegistry { if (profile.provider === 'paypay-bank') { return { from: dashedDate(from), to: dashedDate(to), kinds: ['transaction'] as const } } + if (profile.provider === 'mobilesuica') { + return { from: from.toISOString(), to: to.toISOString(), kinds: ['transaction'] as const } + } return { kinds: ['transaction'] as const } } diff --git a/docs/docs/providers/mobilesuica.md b/docs/docs/providers/mobilesuica.md index 6592c09..7b565ef 100644 --- a/docs/docs/providers/mobilesuica.md +++ b/docs/docs/providers/mobilesuica.md @@ -62,11 +62,26 @@ try { `@repo/capsolve-sp` is a private workspace package in this monorepo. Its model is downloaded by the root `bun install`; set `CAPSOLVE_MODEL_PATH` only when the model is stored somewhere other than the default location. A recognition error causes `login()` to reject rather than prompting for a fallback answer. +### CAPTCHA solver + +The bundled `@repo/capsolve-sp` package recognizes the five-character CAPTCHA image returned by Mobile Suica. Run `bun install` at the repository root before starting the server; it downloads the ONNX model used by the solver. + +The default model path is `../../models/model.onnx`, resolved from the process working directory. When running the SDK from another directory or when the model is stored elsewhere, provide its absolute path with `CAPSOLVE_MODEL_PATH`: + +```ini +CAPSOLVE_MODEL_PATH=/absolute/path/to/model.onnx +``` + +For direct SDK use, create one solver and pass its `solve` method to `onCaptcha`, as in the example above. In the Mnie application this is already wired into the Mobile Suica login flow: it shows the CAPTCHA and fills a suggested answer. The account holder can review or replace that answer before submitting it. + `login()` reads `MOBILE_SUICA_USER` and `MOBILE_SUICA_PASS` when `user` and `password` are not supplied. CAPTCHA answers are not logged or retained by the client. ## Returned values -`getUsageHistory()` returns the 100 SF history rows displayed by Mobile Suica. +`getUsageHistory()` returns the SF history rows currently displayed by Mobile Suica. The website +does not expose pagination, but it supports searching for records on or before a selected date. +The provider repeats that search for `from` / `to` range requests. If a single day reaches the +100-record display limit, the provider throws instead of returning incomplete history. | Property | Description | | --------- | --------------------------------------------- | diff --git a/packages/provider-mobile-suica/README.md b/packages/provider-mobile-suica/README.md index f36ed81..b2b1312 100644 --- a/packages/provider-mobile-suica/README.md +++ b/packages/provider-mobile-suica/README.md @@ -1,6 +1,8 @@ # client-mobilesuica -Mobile Suica web client for Mnie. It only reads SF (electronic money) usage history. +Mobile Suica web client for Mnie. It reads SF (electronic money) usage history. Mobile Suica has +no pagination, but supports searching for records on or before a selected date; range requests +repeat that search to retrieve older history. ```ts import { login } from '@mnie/provider-mobile-suica' diff --git a/packages/provider-mobile-suica/src/index.ts b/packages/provider-mobile-suica/src/index.ts index a48d0c0..adbd888 100644 --- a/packages/provider-mobile-suica/src/index.ts +++ b/packages/provider-mobile-suica/src/index.ts @@ -75,7 +75,12 @@ export interface MobileSuicaProfile { logout(): Promise } -const historyReaders = new WeakMap Promise>() +const mobileSuicaHistoryPageLimit = 100 + +const historyReaders = new WeakMap< + MobileSuicaProfile, + (onOrBefore?: Date) => Promise +>() class MobileSuicaCaptchaRequiredError extends Error {} @@ -117,18 +122,16 @@ export const createProvider = ( invoke: async (name, request) => { if (name === 'accounts.list') return { items: [account] } as Page as never if (name === 'transactions.list') { - const readHistory = historyReaders.get(profile) - if (!readHistory) throw new Error('Mobile Suica profile does not have a transaction reader') - const items = (await readHistory()).map((row) => transactionFromHistoryRow(account.id, row)) + const input = (request ?? {}) as DateRangeRequest + const items = await transactionsForRequest(profile, account, input) return { items } as Page as never } if (name === 'history.list') { - const input = request as HistoryListRequest + const input = (request ?? {}) as HistoryListRequest if (input.kinds?.some((kind) => kind !== 'transaction')) { throw new Error('Mobile Suica history.list supports transaction history only') } - const transactions = (await createProvider(profile).invoke('transactions.list', input)) - .items + const transactions = await transactionsForRequest(profile, account, input) return { items: transactions.map((transaction) => ({ kind: 'transaction' as const, @@ -144,6 +147,87 @@ export const createProvider = ( } } +const validHistoryDate = (value: string | undefined, name: 'from' | 'to') => { + if (value === undefined) return undefined + const date = new Date(value) + if (!Number.isFinite(date.getTime())) throw new Error(`Mobile Suica ${name} must be a valid date`) + return date +} + +const transactionsForRequest = async ( + profile: MobileSuicaProfile, + account: Account, + request: DateRangeRequest, +) => { + if (request.accountId && request.accountId !== account.id) return [] + if (request.cursor) { + throw new Error('Mobile Suica transaction history does not support cursors') + } + if (request.limit !== undefined && (!Number.isSafeInteger(request.limit) || request.limit < 0)) { + throw new Error('Mobile Suica transaction history limit must be a non-negative safe integer') + } + const from = validHistoryDate(request.from, 'from') + const to = validHistoryDate(request.to, 'to') + if (from && to && from > to) throw new Error('Mobile Suica from must not be after to') + + const readHistory = historyReaders.get(profile) + if (!readHistory) throw new Error('Mobile Suica profile does not have a transaction reader') + const fromDay = from && startOfTokyoDay(from) + const toDay = to ? endOfTokyoDay(to) : new Date() + const rows: ParsedHistoryTableRow[] = [] + let cursor = toDay + while (true) { + const page = await readHistory(cursor) + if (page.length === 0) break + const oldest = oldestHistoryDate(page) + rows.push(...page) + + if (request.limit !== undefined && rows.length >= request.limit) break + if (page.length < mobileSuicaHistoryPageLimit) break + if (fromDay && oldest <= fromDay) { + if (oldest.getTime() === fromDay.getTime()) { + throw new Error( + `Mobile Suica history reached the ${mobileSuicaHistoryPageLimit}-record limit on ${tokyoDate(oldest)}; the requested range cannot be retrieved completely`, + ) + } + break + } + cursor = dayBeforeTokyo(oldest) + } + + const transactions = rows.map((row) => transactionFromHistoryRow(account.id, row)) + const filtered = transactions.filter((transaction) => { + const occurredAt = new Date(transaction.occurredAt) + return (!fromDay || occurredAt >= fromDay) && occurredAt <= toDay + }) + const unique = [...new Map(filtered.map((transaction) => [transaction.id, transaction])).values()] + unique.sort((left, right) => right.occurredAt.localeCompare(left.occurredAt)) + return request.limit === undefined ? unique : unique.slice(0, request.limit) +} + +const tokyoOffsetMs = 9 * 60 * 60_000 + +const tokyoDate = (date: Date) => { + const shifted = new Date(date.getTime() + tokyoOffsetMs) + return `${shifted.getUTCFullYear()}-${String(shifted.getUTCMonth() + 1).padStart(2, '0')}-${String(shifted.getUTCDate()).padStart(2, '0')}` +} + +const startOfTokyoDay = (date: Date) => new Date(`${tokyoDate(date)}T00:00:00+09:00`) +const endOfTokyoDay = (date: Date) => new Date(`${tokyoDate(date)}T23:59:59.999+09:00`) +const dayBeforeTokyo = (date: Date) => new Date(startOfTokyoDay(date).getTime() - 24 * 60 * 60_000) + +const oldestHistoryDate = (records: ParsedHistoryTableRow[]) => { + const oldest = records.reduce((current, record) => { + const date = new Date(record.date) + if (!Number.isFinite(date.getTime())) { + throw new Error(`Mobile Suica returned an invalid history date: ${record.date}`) + } + return !current || date < current ? date : current + }, undefined) + if (!oldest) throw new Error('Mobile Suica history page had no records') + return oldest +} + export interface MobileSuicaSession { baseURL: string user: string @@ -346,6 +430,7 @@ const tableRows = (html: string) => const normalizeHistoryDates = (records: ParsedHistoryTableRow[], now = new Date()) => { let inferredYear = now.getFullYear() let previousTime = Number.POSITIVE_INFINITY + const occurrences = new Map() return records.map((record) => { const match = record.date.trim().match(/^(?:(\d{4})[年/.-])?(\d{1,2})[月/.-](\d{1,2})日?$/) if (!match) throw new Error(`unsupported Mobile Suica history date: ${record.date}`) @@ -367,11 +452,29 @@ const normalizeHistoryDates = (records: ParsedHistoryTableRow[], now = new Date( } previousTime = time const isoDate = `${String(year).padStart(4, '0')}-${String(month).padStart(2, '0')}-${String(day).padStart(2, '0')}T00:00:00+09:00` - return { ...record, date: isoDate } + const fingerprint = stableHistoryId(record, isoDate) + const occurrence = occurrences.get(fingerprint) ?? 0 + occurrences.set(fingerprint, occurrence + 1) + return { ...record, date: isoDate, id: `${fingerprint}:${occurrence}` } }) } -const parseMobileSuicaUsageHistory = (html: string): ParsedHistoryTableRow[] => { +const stableHistoryId = (record: ParsedHistoryTableRowBase, date: string) => + JSON.stringify([ + date, + record.typeFrom, + record.placeFrom, + record.typeTo, + record.placeTo, + record.balanceText, + record.amountText, + ]) + +const parseMobileSuicaUsageHistory = ( + html: string, + referenceDate = new Date(), + allowEmpty = false, +): ParsedHistoryTableRow[] => { const records: ParsedHistoryTableRow[] = [] for (const cells of tableRows(html)) { // The first cell is the print-selection checkbox. The 100 history rows in @@ -409,9 +512,10 @@ const parseMobileSuicaUsageHistory = (html: string): ParsedHistoryTableRow[] => return classes.includes('loginBtn') && classes.includes('ButtonBox') }) if (hasLoginButton) throw new MobileSuicaCaptchaRequiredError('logged out, CAPTCHA is needed') + if (allowEmpty) return [] throw new Error('usage history page did not include any usage rows') } - return normalizeHistoryDates(records) + return normalizeHistoryDates(records, referenceDate) } const yenAmount = (value: number | null) => @@ -491,6 +595,18 @@ const submit = async ( const formBody = (fields: Record) => new URLSearchParams(fields).toString() +const historySearchFields = (html: string, onOrBefore: Date) => { + const baseVariable = inputFields(html).baseVariable + if (!baseVariable) throw new Error('Mobile Suica history page did not include baseVariable') + const [year, month, day] = tokyoDate(onOrBefore).split('-') + return { + baseVariable, + specifyYearMonth: `${year}/${month}`, + specifyDay: day!, + SEARCH: '検索', + } +} + export const login = async (options: MobileSuicaLoginOptions): Promise => { const baseURL = normalizeMobileSuicaOrigin(options.baseURL) const { user, password } = readCredentials(options) @@ -602,9 +718,17 @@ const createProfile = ( if (!response.ok) throw new Error(`logout request failed: HTTP ${response.status}`) }, } - historyReaders.set(profile, async () => { + historyReaders.set(profile, async (onOrBefore) => { const html = await submit(historyUrl, {}, jar, historyUrl, 'usage history request') - return parseMobileSuicaUsageHistory(html) + if (!onOrBefore) return parseMobileSuicaUsageHistory(html) + const result = await submit( + historyUrl, + historySearchFields(html, onOrBefore), + jar, + historyUrl, + 'usage history search request', + ) + return parseMobileSuicaUsageHistory(result, onOrBefore, true) }) return profile } @@ -627,6 +751,7 @@ export const importSession = async (session: MobileSuicaSession): Promise Date: Tue, 14 Jul 2026 22:46:06 +0900 Subject: [PATCH 2/7] add reconciliation foundation --- apps/mnie-server/src/app.ts | 2 +- apps/mnie-server/src/cron/index.ts | 22 +- apps/mnie-server/src/db/index.ts | 191 ++++++ apps/mnie-server/src/db/schema.ts | 157 +++++ apps/mnie-server/src/history.ts | 34 +- apps/mnie-server/src/observations.test.ts | 66 ++ apps/mnie-server/src/observations.ts | 320 +++++++++ apps/mnie-server/src/providers/registry.ts | 16 +- apps/mnie-server/src/reconciliation.test.ts | 135 ++++ apps/mnie-server/src/reconciliation.ts | 613 ++++++++++++++++++ apps/mnie-server/src/rpc/handler.ts | 19 +- apps/mnie-server/src/rpc/workspace.ts | 44 +- .../src/security/oauth-provider.ts | 2 +- apps/mnie-ui/src/App.vue | 3 + apps/mnie-ui/src/api.ts | 45 ++ .../src/components/ApiKeyPolicyEditor.vue | 2 +- apps/mnie-ui/src/constants/provider.ts | 2 +- .../reconciliation/ReconciliationView.vue | 169 +++++ .../src/features/settings/SettingsView.vue | 16 +- package.json | 2 +- packages/mnie-types/src/financial.ts | 3 +- packages/mnie-types/src/index.ts | 2 + packages/mnie-types/src/observations.ts | 65 ++ packages/mnie-types/src/reconciliation.ts | 177 +++++ vite.config.ts | 2 + 25 files changed, 2074 insertions(+), 35 deletions(-) create mode 100644 apps/mnie-server/src/observations.test.ts create mode 100644 apps/mnie-server/src/observations.ts create mode 100644 apps/mnie-server/src/reconciliation.test.ts create mode 100644 apps/mnie-server/src/reconciliation.ts create mode 100644 apps/mnie-ui/src/features/reconciliation/ReconciliationView.vue create mode 100644 packages/mnie-types/src/observations.ts create mode 100644 packages/mnie-types/src/reconciliation.ts diff --git a/apps/mnie-server/src/app.ts b/apps/mnie-server/src/app.ts index aac04af..d7611cc 100644 --- a/apps/mnie-server/src/app.ts +++ b/apps/mnie-server/src/app.ts @@ -56,7 +56,7 @@ export const createServerApp = ( baseUrl: new URL(config.origin), resourceServerUrl: new URL('/api/ws', config.origin), resourceName: 'Mnie finance management', - scopesSupported: ['read', 'trade'], + scopesSupported: ['read', 'trade', 'reconcile'], provider: oauthProvider, authorizationOptions: { rateLimit: false }, tokenOptions: { rateLimit: false }, diff --git a/apps/mnie-server/src/cron/index.ts b/apps/mnie-server/src/cron/index.ts index 4813809..1e8e04b 100644 --- a/apps/mnie-server/src/cron/index.ts +++ b/apps/mnie-server/src/cron/index.ts @@ -3,6 +3,7 @@ import type { Db } from '../db' import { fetchAssetValuation, saveAssetValuation } from '../assets' import { checkProfileAvailability, listProfiles, type CachedAvailability } from '../availability' import type { ProviderRegistry } from '../providers/registry' +import { runQueuedReconciliation } from '../reconciliation' export interface CronJobStatus { id: string @@ -69,6 +70,12 @@ export const createCronSystem = ( schedule: '* * * * *', running: false, } + const reconciliationStatus: CronJobStatus = { + id: 'reconciliation', + label: '取引照合', + schedule: '* * * * *', + running: false, + } const lastSessionRuns = new Map() const lastAssetRuns = new Map() const activeRuns = new Set>() @@ -129,6 +136,12 @@ export const createCronSystem = ( if (errors.length) throw new Error(errors.join('; ')) } + const reconcile = async () => { + while (await runQueuedReconciliation(db)) { + // Drain queued ranges before the next cron interval. + } + } + const run = (status: CronJobStatus, operation: () => Promise) => { if (closed) return const active = runExclusive(status, operation) @@ -142,15 +155,22 @@ export const createCronSystem = ( Bun.cron(availabilityStatus.schedule, () => run(availabilityStatus, refreshAvailability)), Bun.cron(sessionStatus.schedule, () => run(sessionStatus, maintainSessions)), Bun.cron(assetStatus.schedule, () => run(assetStatus, updateAssets)), + Bun.cron(reconciliationStatus.schedule, () => run(reconciliationStatus, reconcile)), ] if (options.start !== false) { run(availabilityStatus, refreshAvailability) run(sessionStatus, maintainSessions) run(assetStatus, updateAssets) + run(reconciliationStatus, reconcile) } return { - jobs: () => [{ ...sessionStatus }, { ...availabilityStatus }, { ...assetStatus }], + jobs: () => [ + { ...sessionStatus }, + { ...availabilityStatus }, + { ...assetStatus }, + { ...reconciliationStatus }, + ], availability: (profileId) => Object.fromEntries([...availabilityCache].filter(([id]) => !profileId || id === profileId)), setAvailability: (profileId, value) => availabilityCache.set(profileId, value), diff --git a/apps/mnie-server/src/db/index.ts b/apps/mnie-server/src/db/index.ts index 4e7d79f..9119c53 100644 --- a/apps/mnie-server/src/db/index.ts +++ b/apps/mnie-server/src/db/index.ts @@ -85,6 +85,7 @@ export const createDb = (path: string) => { .map((column) => column.name) if (!accountProfileColumns.includes('color')) sqlite.run('ALTER TABLE account_profiles ADD COLUMN color TEXT') + sqlite.run(`UPDATE account_profiles SET provider = 'mobile-suica' WHERE provider = 'mobilesuica'`) sqlite.run(` CREATE TABLE IF NOT EXISTS asset_valuations ( id INTEGER PRIMARY KEY AUTOINCREMENT, @@ -131,6 +132,196 @@ export const createDb = (path: string) => { fetched_at INTEGER NOT NULL ) `) + sqlite.run(` + CREATE TABLE IF NOT EXISTS financial_accounts ( + id TEXT PRIMARY KEY, + profile_id TEXT NOT NULL, + connector_type_id TEXT NOT NULL, + institution_id TEXT NOT NULL, + provider_account_id TEXT NOT NULL, + kind TEXT NOT NULL, + created_at INTEGER NOT NULL, + updated_at INTEGER NOT NULL, + UNIQUE (profile_id, provider_account_id) + ) + `) + sqlite.run(` + CREATE TABLE IF NOT EXISTS transaction_observation_snapshots ( + id TEXT PRIMARY KEY, + profile_id TEXT NOT NULL, + connector_type_id TEXT NOT NULL, + payload_json TEXT NOT NULL, + fetched_at INTEGER NOT NULL + ) + `) + sqlite.run(` + CREATE INDEX IF NOT EXISTS transaction_observation_snapshots_profile_fetched_idx + ON transaction_observation_snapshots (profile_id, fetched_at DESC) + `) + sqlite.run(` + CREATE TABLE IF NOT EXISTS transaction_observations ( + id TEXT PRIMARY KEY, + profile_id TEXT NOT NULL, + account_id TEXT NOT NULL, + connector_type_id TEXT NOT NULL, + institution_id TEXT NOT NULL, + provider_account_id TEXT NOT NULL, + provider_transaction_id TEXT, + fingerprint TEXT NOT NULL, + current_revision INTEGER NOT NULL, + first_seen_at INTEGER NOT NULL, + last_seen_at INTEGER NOT NULL + ) + `) + sqlite.run(` + CREATE UNIQUE INDEX IF NOT EXISTS transaction_observations_profile_upstream_unique + ON transaction_observations (profile_id, provider_transaction_id) + WHERE provider_transaction_id IS NOT NULL + `) + sqlite.run(` + CREATE INDEX IF NOT EXISTS transaction_observations_profile_fingerprint_idx + ON transaction_observations (profile_id, account_id, fingerprint) + `) + sqlite.run(` + CREATE TABLE IF NOT EXISTS transaction_observation_revisions ( + observation_id TEXT NOT NULL, + revision INTEGER NOT NULL, + snapshot_id TEXT NOT NULL, + normalized_json TEXT NOT NULL, + parser_version TEXT NOT NULL, + fetched_at INTEGER NOT NULL, + PRIMARY KEY (observation_id, revision) + ) + `) + sqlite.run(` + CREATE INDEX IF NOT EXISTS transaction_observation_revisions_snapshot_idx + ON transaction_observation_revisions (snapshot_id) + `) + sqlite.run(` + CREATE TABLE IF NOT EXISTS ledger_accounts ( + id TEXT PRIMARY KEY, + class TEXT NOT NULL, + financial_account_id TEXT, + name TEXT NOT NULL, + created_at INTEGER NOT NULL + ) + `) + sqlite.run(` + CREATE UNIQUE INDEX IF NOT EXISTS ledger_accounts_financial_account_unique + ON ledger_accounts (financial_account_id) WHERE financial_account_id IS NOT NULL + `) + sqlite.run(` + CREATE TABLE IF NOT EXISTS economic_events ( + id TEXT PRIMARY KEY, + kind TEXT NOT NULL, + state TEXT NOT NULL, + completeness TEXT NOT NULL, + occurred_from INTEGER NOT NULL, + occurred_to INTEGER NOT NULL, + metadata_json TEXT, + created_at INTEGER NOT NULL, + updated_at INTEGER NOT NULL + ) + `) + sqlite.run(` + CREATE INDEX IF NOT EXISTS economic_events_occurred_idx + ON economic_events (occurred_from, occurred_to) + `) + sqlite.run(` + CREATE TABLE IF NOT EXISTS event_postings ( + id TEXT PRIMARY KEY, + event_id TEXT NOT NULL, + ledger_account_id TEXT NOT NULL, + side TEXT NOT NULL, + amount_json TEXT NOT NULL, + role TEXT + ) + `) + sqlite.run(` + CREATE TABLE IF NOT EXISTS observation_bindings ( + id TEXT PRIMARY KEY, + observation_id TEXT NOT NULL, + event_id TEXT NOT NULL, + posting_ids_json TEXT, + state TEXT NOT NULL, + provenance TEXT NOT NULL, + confidence TEXT, + matcher_version TEXT, + evidence_json TEXT NOT NULL, + created_at INTEGER NOT NULL, + updated_at INTEGER NOT NULL + ) + `) + sqlite.run(` + CREATE UNIQUE INDEX IF NOT EXISTS observation_bindings_confirmed_observation_unique + ON observation_bindings (observation_id) WHERE state = 'confirmed' + `) + sqlite.run(` + CREATE TABLE IF NOT EXISTS event_relations ( + id TEXT PRIMARY KEY, + from_event_id TEXT NOT NULL, + to_event_id TEXT NOT NULL, + type TEXT NOT NULL, + created_at INTEGER NOT NULL + ) + `) + sqlite.run(` + CREATE TABLE IF NOT EXISTS account_links ( + id TEXT PRIMARY KEY, + source_account_id TEXT NOT NULL, + target_account_id TEXT NOT NULL, + type TEXT NOT NULL, + instrument_json TEXT, + valid_from INTEGER, + valid_to INTEGER, + source TEXT NOT NULL, + confirmed INTEGER NOT NULL + ) + `) + sqlite.run(` + CREATE TABLE IF NOT EXISTS reconciliation_proposals ( + id TEXT PRIMARY KEY, + candidate_key TEXT NOT NULL, + event_id TEXT NOT NULL, + score TEXT NOT NULL, + state TEXT NOT NULL, + created_at INTEGER NOT NULL + ) + `) + sqlite.run(` + CREATE UNIQUE INDEX IF NOT EXISTS reconciliation_proposals_candidate_unique + ON reconciliation_proposals (candidate_key) WHERE state = 'proposed' + `) + sqlite.run(` + CREATE TABLE IF NOT EXISTS reconciliation_decisions ( + id TEXT PRIMARY KEY, + proposal_id TEXT NOT NULL, + candidate_key TEXT NOT NULL, + decision TEXT NOT NULL, + reason TEXT, + decided_at INTEGER NOT NULL + ) + `) + sqlite.run(` + CREATE INDEX IF NOT EXISTS reconciliation_decisions_candidate_idx + ON reconciliation_decisions (candidate_key, decided_at DESC) + `) + sqlite.run(` + CREATE TABLE IF NOT EXISTS reconciliation_jobs ( + id TEXT PRIMARY KEY, + from_at INTEGER NOT NULL, + to_at INTEGER NOT NULL, + state TEXT NOT NULL, + created_at INTEGER NOT NULL, + started_at INTEGER, + completed_at INTEGER, + error TEXT + ) + `) + sqlite.run(` + CREATE INDEX IF NOT EXISTS reconciliation_jobs_state_created_idx + ON reconciliation_jobs (state, created_at) + `) sqlite.run(` INSERT OR IGNORE INTO account_profiles (id, provider, label, keyring_account, created_at, updated_at) SELECT id, 'sbisec', label, keyring_account, created_at, updated_at FROM sbi_passkeys diff --git a/apps/mnie-server/src/db/schema.ts b/apps/mnie-server/src/db/schema.ts index ce10eca..87210f4 100644 --- a/apps/mnie-server/src/db/schema.ts +++ b/apps/mnie-server/src/db/schema.ts @@ -79,6 +79,163 @@ export const historySyncs = sqliteTable('history_syncs', { fetchedAt: integer('fetched_at', { mode: 'timestamp_ms' }).notNull(), }) +export const financialAccounts = sqliteTable( + 'financial_accounts', + { + id: text('id').primaryKey(), + profileId: text('profile_id').notNull(), + connectorTypeId: text('connector_type_id').notNull(), + institutionId: text('institution_id').notNull(), + providerAccountId: text('provider_account_id').notNull(), + kind: text('kind').notNull(), + createdAt: integer('created_at', { mode: 'timestamp_ms' }).notNull(), + updatedAt: integer('updated_at', { mode: 'timestamp_ms' }).notNull(), + }, + (table) => [ + uniqueIndex('financial_accounts_profile_provider_account_unique').on( + table.profileId, + table.providerAccountId, + ), + ], +) + +/** A fetched, normalized history page. Raw provider payloads are added separately. */ +export const transactionObservationSnapshots = sqliteTable('transaction_observation_snapshots', { + id: text('id').primaryKey(), + profileId: text('profile_id').notNull(), + connectorTypeId: text('connector_type_id').notNull(), + payload: text('payload_json', { mode: 'json' }).$type().notNull(), + fetchedAt: integer('fetched_at', { mode: 'timestamp_ms' }).notNull(), +}) + +export const transactionObservations = sqliteTable( + 'transaction_observations', + { + id: text('id').primaryKey(), + profileId: text('profile_id').notNull(), + accountId: text('account_id').notNull(), + connectorTypeId: text('connector_type_id').notNull(), + institutionId: text('institution_id').notNull(), + providerAccountId: text('provider_account_id').notNull(), + providerTransactionId: text('provider_transaction_id'), + fingerprint: text('fingerprint').notNull(), + currentRevision: integer('current_revision').notNull(), + firstSeenAt: integer('first_seen_at', { mode: 'timestamp_ms' }).notNull(), + lastSeenAt: integer('last_seen_at', { mode: 'timestamp_ms' }).notNull(), + }, + (table) => [ + uniqueIndex('transaction_observations_profile_upstream_unique').on( + table.profileId, + table.providerTransactionId, + ), + ], +) + +export const transactionObservationRevisions = sqliteTable( + 'transaction_observation_revisions', + { + observationId: text('observation_id').notNull(), + revision: integer('revision').notNull(), + snapshotId: text('snapshot_id').notNull(), + normalized: text('normalized_json', { mode: 'json' }).$type().notNull(), + parserVersion: text('parser_version').notNull(), + fetchedAt: integer('fetched_at', { mode: 'timestamp_ms' }).notNull(), + }, + (table) => [primaryKey({ columns: [table.observationId, table.revision] })], +) + +export const ledgerAccounts = sqliteTable('ledger_accounts', { + id: text('id').primaryKey(), + class: text('class').notNull(), + financialAccountId: text('financial_account_id'), + name: text('name').notNull(), + createdAt: integer('created_at', { mode: 'timestamp_ms' }).notNull(), +}) + +export const economicEvents = sqliteTable('economic_events', { + id: text('id').primaryKey(), + kind: text('kind').notNull(), + state: text('state').notNull(), + completeness: text('completeness').notNull(), + occurredFrom: integer('occurred_from', { mode: 'timestamp_ms' }).notNull(), + occurredTo: integer('occurred_to', { mode: 'timestamp_ms' }).notNull(), + metadata: text('metadata_json', { mode: 'json' }).$type(), + createdAt: integer('created_at', { mode: 'timestamp_ms' }).notNull(), + updatedAt: integer('updated_at', { mode: 'timestamp_ms' }).notNull(), +}) + +export const eventPostings = sqliteTable('event_postings', { + id: text('id').primaryKey(), + eventId: text('event_id').notNull(), + ledgerAccountId: text('ledger_account_id').notNull(), + side: text('side').notNull(), + amount: text('amount_json', { mode: 'json' }).$type().notNull(), + role: text('role'), +}) + +export const observationBindings = sqliteTable('observation_bindings', { + id: text('id').primaryKey(), + observationId: text('observation_id').notNull(), + eventId: text('event_id').notNull(), + postingIds: text('posting_ids_json', { mode: 'json' }).$type(), + state: text('state').notNull(), + provenance: text('provenance').notNull(), + confidence: text('confidence'), + matcherVersion: text('matcher_version'), + evidence: text('evidence_json', { mode: 'json' }).$type().notNull(), + createdAt: integer('created_at', { mode: 'timestamp_ms' }).notNull(), + updatedAt: integer('updated_at', { mode: 'timestamp_ms' }).notNull(), +}) + +export const eventRelations = sqliteTable('event_relations', { + id: text('id').primaryKey(), + fromEventId: text('from_event_id').notNull(), + toEventId: text('to_event_id').notNull(), + type: text('type').notNull(), + createdAt: integer('created_at', { mode: 'timestamp_ms' }).notNull(), +}) + +export const accountLinks = sqliteTable('account_links', { + id: text('id').primaryKey(), + sourceAccountId: text('source_account_id').notNull(), + targetAccountId: text('target_account_id').notNull(), + type: text('type').notNull(), + instrument: text('instrument_json', { mode: 'json' }).$type(), + validFrom: integer('valid_from', { mode: 'timestamp_ms' }), + validTo: integer('valid_to', { mode: 'timestamp_ms' }), + source: text('source').notNull(), + confirmed: integer('confirmed', { mode: 'boolean' }).notNull(), +}) + +export const reconciliationProposals = sqliteTable('reconciliation_proposals', { + id: text('id').primaryKey(), + candidateKey: text('candidate_key').notNull(), + eventId: text('event_id').notNull(), + score: text('score').notNull(), + state: text('state').notNull(), + createdAt: integer('created_at', { mode: 'timestamp_ms' }).notNull(), +}) + +export const reconciliationDecisions = sqliteTable('reconciliation_decisions', { + id: text('id').primaryKey(), + proposalId: text('proposal_id').notNull(), + candidateKey: text('candidate_key').notNull(), + decision: text('decision').notNull(), + reason: text('reason'), + decidedAt: integer('decided_at', { mode: 'timestamp_ms' }).notNull(), +}) + +export const reconciliationJobs = sqliteTable('reconciliation_jobs', { + id: text('id').primaryKey(), + from: integer('from_at', { mode: 'timestamp_ms' }).notNull(), + to: integer('to_at', { mode: 'timestamp_ms' }).notNull(), + state: text('state').notNull(), + createdAt: integer('created_at', { mode: 'timestamp_ms' }).notNull(), + startedAt: integer('started_at', { mode: 'timestamp_ms' }), + completedAt: integer('completed_at', { mode: 'timestamp_ms' }), + error: text('error'), +}) + export const apiKeys = sqliteTable( 'api_keys', { diff --git a/apps/mnie-server/src/history.ts b/apps/mnie-server/src/history.ts index cf7d30c..641a0fa 100644 --- a/apps/mnie-server/src/history.ts +++ b/apps/mnie-server/src/history.ts @@ -2,7 +2,9 @@ import { and, asc, eq, gte, inArray, lte } from 'drizzle-orm' import type { HistoryItem, HistoryListRequest, Transaction } from '@mnie/types' import type { Db } from './db' import { accountProfiles, assetValuations, historySyncs, historyTransactions } from './db/schema' +import { persistTransactionObservations } from './observations' import type { ProviderRegistry } from './providers/registry' +import { enqueueReconciliation } from './reconciliation' const refreshIntervalMs = 5 * 60 * 60_000 const defaultRangeMs = 30 * 24 * 60 * 60_000 @@ -59,26 +61,43 @@ const syncTransactions = async ( } const items = await fetchPage(from, to) const fetchedAt = new Date() - for (const item of items) { - if (item.kind !== 'transaction') continue - const occurredAt = new Date(item.occurredAt) + const transactions = items + .filter( + (item): item is Extract => + item.kind === 'transaction', + ) + .map((item) => item.transaction) + const uniqueTransactions = + profile.provider === 'mobile-suica' + ? transactions + : [...new Map(transactions.map((transaction) => [transaction.id, transaction])).values()] + const observations = await persistTransactionObservations( + db, + { id: profile.id, provider: profile.provider }, + uniqueTransactions, + fetchedAt, + ) + for (const { historyTransaction } of observations) { + const occurredAt = new Date(historyTransaction.occurredAt) if (!Number.isFinite(occurredAt.getTime())) { - throw new Error(`provider returned invalid transaction date: ${item.occurredAt}`) + throw new Error( + `provider returned invalid transaction date: ${historyTransaction.occurredAt}`, + ) } await db .insert(historyTransactions) .values({ profileId: profile.id, - transactionId: item.transaction.id, + transactionId: historyTransaction.id, occurredAt, - transaction: item.transaction, + transaction: historyTransaction, fetchedAt, }) .onConflictDoUpdate({ target: [historyTransactions.profileId, historyTransactions.transactionId], set: { occurredAt, - transaction: item.transaction, + transaction: historyTransaction, fetchedAt, }, }) @@ -94,6 +113,7 @@ const syncTransactions = async ( fetchedAt, }, }) + if (observations.length) await enqueueReconciliation(db, from, to) }, { forceLogin }, ) diff --git a/apps/mnie-server/src/observations.test.ts b/apps/mnie-server/src/observations.test.ts new file mode 100644 index 0000000..519f31e --- /dev/null +++ b/apps/mnie-server/src/observations.test.ts @@ -0,0 +1,66 @@ +import { describe, expect, test } from 'bun:test' +import type { Transaction } from '@mnie/types' +import { createDb } from './db' +import { transactionObservationRevisions } from './db/schema' +import { persistTransactionObservations } from './observations' + +const mobileTransaction = (description: string, balance: string): Transaction => ({ + id: `source-${description}`, + accountId: 'mobile-suica', + kind: 'charge', + direction: 'credit', + status: 'posted', + amount: { kind: 'money', money: { currency: 'JPY', value: '5000' } }, + occurredAt: '2026-07-14T00:00:00+09:00', + description, + balanceAfter: { kind: 'money', money: { currency: 'JPY', value: balance } }, + charge: { method: 'card' }, +}) + +const smbcTransaction = (status: Transaction['status'] = 'posted'): Transaction => ({ + id: 'meisai-1', + accountId: '0001-1234567', + kind: 'withdrawal', + direction: 'debit', + status, + amount: { kind: 'money', money: { currency: 'JPY', value: '5000' } }, + occurredAt: '2026-07-14T12:00:00+09:00', + description: 'モバイルSuica', +}) + +describe('transaction observations', () => { + test('keeps Mobile Suica observation identities when a new row is prepended', async () => { + const db = createDb(':memory:') + const profile = { id: 'suica', provider: 'mobile-suica' } + const initial = await persistTransactionObservations(db, profile, [ + mobileTransaction('older-a', '10000'), + mobileTransaction('older-b', '5000'), + ]) + const next = await persistTransactionObservations(db, profile, [ + mobileTransaction('newer', '15000'), + mobileTransaction('older-a', '10000'), + mobileTransaction('older-b', '5000'), + ]) + + expect(next[1]!.observation.id).toBe(initial[0]!.observation.id) + expect(next[2]!.observation.id).toBe(initial[1]!.observation.id) + expect(next[0]!.historyTransaction.id).toBe(next[0]!.observation.id) + expect(next[1]!.observation.timestamps.precision).toBe('day') + }) + + test('creates a new revision only when a stable upstream observation changes', async () => { + const db = createDb(':memory:') + const profile = { id: 'smbc', provider: 'smbc-direct' } + const [initial] = await persistTransactionObservations(db, profile, [smbcTransaction()]) + const [unchanged] = await persistTransactionObservations(db, profile, [smbcTransaction()]) + const [changed] = await persistTransactionObservations(db, profile, [ + smbcTransaction('reversed'), + ]) + const revisions = await db.select().from(transactionObservationRevisions) + + expect(unchanged!.observation.id).toBe(initial!.observation.id) + expect(unchanged!.observation.source.revision).toBe(1) + expect(changed!.observation.source.revision).toBe(2) + expect(revisions).toHaveLength(2) + }) +}) diff --git a/apps/mnie-server/src/observations.ts b/apps/mnie-server/src/observations.ts new file mode 100644 index 0000000..f3fe5a7 --- /dev/null +++ b/apps/mnie-server/src/observations.ts @@ -0,0 +1,320 @@ +import { and, desc, eq } from 'drizzle-orm' +import type { AccountKind, TimePrecision, Transaction, TransactionObservation } from '@mnie/types' +import type { Db } from './db' +import { + financialAccounts, + transactionObservationRevisions, + transactionObservationSnapshots, + transactionObservations, +} from './db/schema' +import { randomId, sha256 } from './security/crypto' + +const parserVersion = 'observation-v1' + +export interface ObservationProfile { + id: string + provider: string +} + +interface SnapshotItem { + observationId: string + fingerprint: string +} + +interface SnapshotPayload { + version: 1 + items: SnapshotItem[] +} + +interface PendingObservation { + transaction: Transaction + fingerprint: string + observationId?: string + providerTransactionId?: string +} + +export interface PersistedObservation { + observation: TransactionObservation + /** The history-compatible transaction; Mobile Suica uses the persistent observation ID. */ + historyTransaction: Transaction +} + +const institutionId = (connectorTypeId: string) => connectorTypeId + +const accountKind = (connectorTypeId: string): AccountKind => { + if (connectorTypeId === 'mobile-suica') return 'transit-card' + if (connectorTypeId === 'smbc-direct' || connectorTypeId === 'paypay-bank') return 'bank' + return 'other' +} + +const precisionFor = (connectorTypeId: string): TimePrecision => + connectorTypeId === 'mobile-suica' ? 'day' : 'instant' + +const hasStableProviderTransactionId = (connectorTypeId: string) => + connectorTypeId !== 'mobile-suica' + +const fingerprintFor = (transaction: Transaction) => + sha256( + JSON.stringify([ + transaction.occurredAt, + transaction.kind, + transaction.direction, + transaction.status, + transaction.amount, + transaction.description, + transaction.balanceAfter, + ]), + ) + +const parseSnapshot = (payload: unknown): SnapshotPayload | undefined => { + if (!payload || typeof payload !== 'object') return undefined + const value = payload as { version?: unknown; items?: unknown } + if (value.version !== 1 || !Array.isArray(value.items)) return undefined + const items = value.items.flatMap((item) => { + if (!item || typeof item !== 'object') return [] + const value = item as { observationId?: unknown; fingerprint?: unknown } + return typeof value.observationId === 'string' && typeof value.fingerprint === 'string' + ? [{ observationId: value.observationId, fingerprint: value.fingerprint }] + : [] + }) + return { version: 1, items } +} + +/** Returns maximum ordered equal-fingerprint matches between two snapshots. */ +const alignSnapshot = (previous: SnapshotItem[], current: PendingObservation[]) => { + const width = current.length + 1 + const scores = new Uint32Array((previous.length + 1) * width) + const at = (row: number, column: number) => row * width + column + for (let row = 1; row <= previous.length; row += 1) { + for (let column = 1; column <= current.length; column += 1) { + scores[at(row, column)] = + previous[row - 1]!.fingerprint === current[column - 1]!.fingerprint + ? scores[at(row - 1, column - 1)]! + 1 + : Math.max(scores[at(row - 1, column)]!, scores[at(row, column - 1)]!) + } + } + const matches = new Map() + let row = previous.length + let column = current.length + while (row > 0 && column > 0) { + if ( + previous[row - 1]!.fingerprint === current[column - 1]!.fingerprint && + scores[at(row, column)] === scores[at(row - 1, column - 1)]! + 1 + ) { + matches.set(column - 1, previous[row - 1]!.observationId) + row -= 1 + column -= 1 + } else if (scores[at(row - 1, column)]! >= scores[at(row, column - 1)]!) { + row -= 1 + } else { + column -= 1 + } + } + return matches +} + +const normalizedObservation = ( + profile: ObservationProfile, + transaction: Transaction, + observationId: string, + fingerprint: string, + revision: number, + firstSeenAt: Date, + lastSeenAt: Date, + providerTransactionId?: string, +): TransactionObservation => { + const connectorTypeId = profile.provider + return { + id: observationId, + profileId: profile.id, + accountId: `financial-account_${sha256(`${profile.id}:${transaction.accountId}`).slice(0, 32)}`, + source: { + connectorTypeId, + institutionId: institutionId(connectorTypeId), + providerAccountId: transaction.accountId, + ...(providerTransactionId ? { providerTransactionId } : {}), + fingerprint, + revision, + firstSeenAt: firstSeenAt.toISOString(), + lastSeenAt: lastSeenAt.toISOString(), + }, + transaction, + timestamps: { occurredAt: transaction.occurredAt, precision: precisionFor(connectorTypeId) }, + } +} + +const ensureFinancialAccount = async ( + db: Db, + profile: ObservationProfile, + transaction: Transaction, + now: Date, +) => { + const id = `financial-account_${sha256(`${profile.id}:${transaction.accountId}`).slice(0, 32)}` + await db + .insert(financialAccounts) + .values({ + id, + profileId: profile.id, + connectorTypeId: profile.provider, + institutionId: institutionId(profile.provider), + providerAccountId: transaction.accountId, + kind: accountKind(profile.provider), + createdAt: now, + updatedAt: now, + }) + .onConflictDoUpdate({ + target: [financialAccounts.profileId, financialAccounts.providerAccountId], + set: { updatedAt: now }, + }) + return id +} + +/** + * Persists provider observations and their immutable normalized revisions. + * It deliberately stores only normalized pages for now; provider raw payload + * capture is added at the provider boundary rather than reconstructed here. + */ +export const persistTransactionObservations = async ( + db: Db, + profile: ObservationProfile, + transactions: Transaction[], + fetchedAt = new Date(), +): Promise => { + const connectorTypeId = profile.provider + const pending: PendingObservation[] = transactions.map((transaction) => ({ + transaction, + fingerprint: fingerprintFor(transaction), + ...(hasStableProviderTransactionId(connectorTypeId) + ? { providerTransactionId: transaction.id } + : {}), + })) + + if (!hasStableProviderTransactionId(connectorTypeId)) { + const [previousSnapshot] = await db + .select() + .from(transactionObservationSnapshots) + .where(eq(transactionObservationSnapshots.profileId, profile.id)) + .orderBy(desc(transactionObservationSnapshots.fetchedAt)) + .limit(1) + const previous = parseSnapshot(previousSnapshot?.payload)?.items ?? [] + const matches = alignSnapshot(previous, pending) + for (const [index, observationId] of matches) pending[index]!.observationId = observationId + } + + for (const item of pending) { + if (item.observationId) continue + if (item.providerTransactionId) { + const [existing] = await db + .select() + .from(transactionObservations) + .where( + and( + eq(transactionObservations.profileId, profile.id), + eq(transactionObservations.providerTransactionId, item.providerTransactionId), + ), + ) + .limit(1) + if (existing) item.observationId = existing.id + } + item.observationId ??= randomId('observation') + } + + const snapshotId = randomId('observation-snapshot') + await db.insert(transactionObservationSnapshots).values({ + id: snapshotId, + profileId: profile.id, + connectorTypeId, + payload: { + version: 1, + items: pending.map((item) => ({ + observationId: item.observationId!, + fingerprint: item.fingerprint, + })), + } satisfies SnapshotPayload, + fetchedAt, + }) + + const result: PersistedObservation[] = [] + for (const item of pending) { + const observationId = item.observationId! + const financialAccountId = await ensureFinancialAccount( + db, + profile, + item.transaction, + fetchedAt, + ) + const [existing] = await db + .select() + .from(transactionObservations) + .where(eq(transactionObservations.id, observationId)) + .limit(1) + const [latestRevision] = await db + .select() + .from(transactionObservationRevisions) + .where(eq(transactionObservationRevisions.observationId, observationId)) + .orderBy(desc(transactionObservationRevisions.revision)) + .limit(1) + const unchanged = + latestRevision && + JSON.stringify(latestRevision.normalized) === JSON.stringify(item.transaction) + const revision = existing + ? unchanged + ? existing.currentRevision + : existing.currentRevision + 1 + : 1 + + if (!existing) { + await db.insert(transactionObservations).values({ + id: observationId, + profileId: profile.id, + accountId: financialAccountId, + connectorTypeId, + institutionId: institutionId(connectorTypeId), + providerAccountId: item.transaction.accountId, + providerTransactionId: item.providerTransactionId ?? null, + fingerprint: item.fingerprint, + currentRevision: revision, + firstSeenAt: fetchedAt, + lastSeenAt: fetchedAt, + }) + } else { + await db + .update(transactionObservations) + .set({ + accountId: financialAccountId, + fingerprint: item.fingerprint, + currentRevision: revision, + lastSeenAt: fetchedAt, + }) + .where(eq(transactionObservations.id, observationId)) + } + if (!unchanged) { + await db.insert(transactionObservationRevisions).values({ + observationId, + revision, + snapshotId, + normalized: item.transaction, + parserVersion, + fetchedAt, + }) + } + const observation = normalizedObservation( + profile, + item.transaction, + observationId, + item.fingerprint, + revision, + existing?.firstSeenAt ?? fetchedAt, + fetchedAt, + item.providerTransactionId, + ) + result.push({ + observation, + historyTransaction: + connectorTypeId === 'mobile-suica' + ? { ...item.transaction, id: observationId } + : item.transaction, + }) + } + return result +} diff --git a/apps/mnie-server/src/providers/registry.ts b/apps/mnie-server/src/providers/registry.ts index 8dc0f90..8cb7b82 100644 --- a/apps/mnie-server/src/providers/registry.ts +++ b/apps/mnie-server/src/providers/registry.ts @@ -188,7 +188,7 @@ export class ProviderRegistry { ], }, { - id: 'mobilesuica', + id: 'mobile-suica', name: 'Mobile Suica', kind: 'transit-card', authentication: 'credentials-and-captcha', @@ -239,7 +239,7 @@ export class ProviderRegistry { ? this.config.authBaseUrl : providerId === 'smbc-direct' ? (this.config.smbcDirectLoginBaseUrl ?? this.config.smbcDirectBaseUrl) - : providerId === 'mobilesuica' + : providerId === 'mobile-suica' ? this.config.mobileSuicaBaseUrl : providerId === 'paypay-bank' ? this.config.payPayBankBaseUrl @@ -426,7 +426,7 @@ export class ProviderRegistry { } } - if (profile.provider === 'mobilesuica') { + if (profile.provider === 'mobile-suica') { const secret = await readSecret(profile.keyringAccount) if (!secret.session) throw new Error('Mobile Suica authentication is required') const imported = await importMobileSuicaSession(secret.session) @@ -599,7 +599,7 @@ export class ProviderRegistry { if (reauthenticateProfileId) { const profile = await this.profile(reauthenticateProfileId) - if (profile.provider !== 'mobilesuica') throw new Error('profile is not Mobile Suica') + if (profile.provider !== 'mobile-suica') throw new Error('profile is not Mobile Suica') const secret = await readSecret(profile.keyringAccount) if (!secret.user || !secret.password) { throw new Error('Mobile Suica credentials are not stored') @@ -614,7 +614,7 @@ export class ProviderRegistry { label = requiredCredential(input, 'label') user = requiredCredential(input, 'user') password = requiredCredential(input, 'password') - profileId = randomId('mobilesuica') + profileId = randomId('mobile-suica') keyringAccount = `profile:${profileId}` createProfile = true } @@ -682,7 +682,7 @@ export class ProviderRegistry { if (pending.createProfile) { await this.db.insert(accountProfiles).values({ id: pending.profileId, - provider: 'mobilesuica', + provider: 'mobile-suica', label: pending.label, keyringAccount: pending.keyringAccount, createdAt: now, @@ -764,7 +764,7 @@ export class ProviderRegistry { } sessionRefreshIntervalMs(profile: AccountProfile) { - return profile.provider === 'smbc-direct' || profile.provider === 'mobilesuica' + return profile.provider === 'smbc-direct' || profile.provider === 'mobile-suica' ? 5 * 60_000 : undefined } @@ -776,7 +776,7 @@ export class ProviderRegistry { if (profile.provider === 'paypay-bank') { return { from: dashedDate(from), to: dashedDate(to), kinds: ['transaction'] as const } } - if (profile.provider === 'mobilesuica') { + if (profile.provider === 'mobile-suica') { return { from: from.toISOString(), to: to.toISOString(), kinds: ['transaction'] as const } } return { kinds: ['transaction'] as const } diff --git a/apps/mnie-server/src/reconciliation.test.ts b/apps/mnie-server/src/reconciliation.test.ts new file mode 100644 index 0000000..e761bf2 --- /dev/null +++ b/apps/mnie-server/src/reconciliation.test.ts @@ -0,0 +1,135 @@ +import { describe, expect, test } from 'bun:test' +import type { Transaction } from '@mnie/types' +import { createDb } from './db' +import { persistTransactionObservations } from './observations' +import { + confirmReconciliationProposal, + enqueueReconciliation, + listAccountLinks, + listEconomicEvents, + listReconciliationProposals, + rejectReconciliationProposal, + runQueuedReconciliation, + upsertAccountLink, +} from './reconciliation' + +const transaction = (id: string, accountId: string): Transaction => ({ + id, + accountId, + kind: 'withdrawal', + direction: 'debit', + status: 'posted', + amount: { kind: 'money', money: { currency: 'JPY', value: '5000' } }, + occurredAt: '2026-07-14T12:00:00+09:00', + description: id, +}) + +const charge = (id: string, accountId: string): Transaction => ({ + id, + accountId, + kind: 'charge', + direction: 'credit', + status: 'posted', + amount: { kind: 'money', money: { currency: 'JPY', value: '5000' } }, + occurredAt: '2026-07-14T00:00:00+09:00', + description: id, + charge: { method: 'card' }, +}) + +describe('reconciliation store', () => { + test('accepts user account links only between persisted financial accounts', async () => { + const db = createDb(':memory:') + const [bank] = await persistTransactionObservations( + db, + { id: 'bank', provider: 'smbc-direct' }, + [transaction('bank-row', 'bank-account')], + ) + const [suica] = await persistTransactionObservations( + db, + { id: 'suica', provider: 'mobile-suica' }, + [transaction('suica-row', 'mobile-suica')], + ) + const saved = await upsertAccountLink(db, { + sourceAccountId: bank!.observation.accountId, + targetAccountId: suica!.observation.accountId, + type: 'funds', + source: 'inferred', + confirmed: false, + }) + + expect(saved.source).toBe('user') + expect(saved.confirmed).toBe(true) + expect(await listAccountLinks(db)).toEqual([saved]) + await expect( + upsertAccountLink(db, { + sourceAccountId: 'missing', + targetAccountId: saved.targetAccountId, + type: 'funds', + source: 'user', + confirmed: true, + }), + ).rejects.toThrow('unknown financial account') + }) + + test('returns an empty event page before a matcher has generated proposals', async () => { + const db = createDb(':memory:') + await expect(listEconomicEvents(db)).resolves.toEqual({ items: [] }) + }) + + test('queues, proposes, confirms, and respects rejected wallet top-ups', async () => { + const db = createDb(':memory:') + const [bank] = await persistTransactionObservations( + db, + { id: 'bank', provider: 'smbc-direct' }, + [transaction('bank-row', 'bank-account')], + ) + const [suica] = await persistTransactionObservations( + db, + { id: 'suica', provider: 'mobile-suica' }, + [charge('suica-row', 'mobile-suica')], + ) + await upsertAccountLink(db, { + sourceAccountId: bank!.observation.accountId, + targetAccountId: suica!.observation.accountId, + type: 'funds', + source: 'user', + confirmed: true, + }) + const from = new Date('2026-07-13T00:00:00+09:00') + const to = new Date('2026-07-15T00:00:00+09:00') + await enqueueReconciliation(db, from, to) + await expect(runQueuedReconciliation(db)).resolves.toBe(1) + const [proposal] = (await listReconciliationProposals(db)).items + expect(proposal?.event.kind).toBe('wallet-topup') + expect(proposal?.bindings).toHaveLength(2) + const confirmed = await confirmReconciliationProposal(db, proposal!.id) + expect(confirmed.event.state).toBe('confirmed') + expect(confirmed.postings.map((posting) => posting.side).sort()).toEqual(['credit', 'debit']) + + // A rejected candidate is retained as an audit decision and never becomes proposed again. + const dbRejected = createDb(':memory:') + const [rejectedBank] = await persistTransactionObservations( + dbRejected, + { id: 'bank', provider: 'smbc-direct' }, + [transaction('bank-row', 'bank-account')], + ) + const [rejectedSuica] = await persistTransactionObservations( + dbRejected, + { id: 'suica', provider: 'mobile-suica' }, + [charge('suica-row', 'mobile-suica')], + ) + await upsertAccountLink(dbRejected, { + sourceAccountId: rejectedBank!.observation.accountId, + targetAccountId: rejectedSuica!.observation.accountId, + type: 'funds', + source: 'user', + confirmed: true, + }) + await enqueueReconciliation(dbRejected, from, to) + await runQueuedReconciliation(dbRejected) + const [rejectedProposal] = (await listReconciliationProposals(dbRejected)).items + await rejectReconciliationProposal(dbRejected, rejectedProposal!.id, 'not related') + await enqueueReconciliation(dbRejected, from, to) + await expect(runQueuedReconciliation(dbRejected)).resolves.toBe(0) + }) +}) diff --git a/apps/mnie-server/src/reconciliation.ts b/apps/mnie-server/src/reconciliation.ts new file mode 100644 index 0000000..3fbc9c9 --- /dev/null +++ b/apps/mnie-server/src/reconciliation.ts @@ -0,0 +1,613 @@ +import { asc, eq } from 'drizzle-orm' +import type { + Amount, + AccountLink, + AccountLinkInput, + EconomicEvent, + EconomicEventView, + EventsListRequest, + EventRelation, + MatchEvidence, + ObservationBinding, + Posting, + ReconciliationProposal, + ReconciliationProposalRequest, + Transaction, + FinancialAccount, +} from '@mnie/types' +import type { Db } from './db' +import { + accountLinks, + economicEvents, + eventPostings, + eventRelations, + financialAccounts, + ledgerAccounts, + observationBindings, + reconciliationDecisions, + reconciliationJobs, + reconciliationProposals, + transactionObservationRevisions, + transactionObservations, +} from './db/schema' +import { randomId } from './security/crypto' + +const asDate = (value: string | undefined, name: string) => { + if (value === undefined) return undefined + const date = new Date(value) + if (!Number.isFinite(date.getTime())) throw new Error(`${name} must be a valid date`) + return date +} + +const asEvent = (row: typeof economicEvents.$inferSelect): EconomicEvent => ({ + id: row.id, + kind: row.kind as EconomicEvent['kind'], + state: row.state as EconomicEvent['state'], + completeness: row.completeness as EconomicEvent['completeness'], + occurredAt: { from: row.occurredFrom.toISOString(), to: row.occurredTo.toISOString() }, + ...(row.metadata ? { metadata: row.metadata as EconomicEvent['metadata'] } : {}), + createdAt: row.createdAt.toISOString(), + updatedAt: row.updatedAt.toISOString(), +}) + +const eventView = async ( + db: Db, + event: typeof economicEvents.$inferSelect, +): Promise => { + const [postings, bindings, relations] = await Promise.all([ + db.select().from(eventPostings).where(eq(eventPostings.eventId, event.id)), + db.select().from(observationBindings).where(eq(observationBindings.eventId, event.id)), + db + .select() + .from(eventRelations) + .where(eq(eventRelations.fromEventId, event.id)) + .orderBy(asc(eventRelations.createdAt)), + ]) + return { + event: asEvent(event), + postings: postings.map( + (posting): Posting => ({ + id: posting.id, + eventId: posting.eventId, + ledgerAccountId: posting.ledgerAccountId, + side: posting.side as Posting['side'], + amount: posting.amount as Posting['amount'], + ...(posting.role ? { role: posting.role as Posting['role'] } : {}), + }), + ), + bindings: bindings.map( + (binding): ObservationBinding => ({ + id: binding.id, + observationId: binding.observationId, + eventId: binding.eventId, + ...(binding.postingIds ? { postingIds: binding.postingIds } : {}), + state: binding.state as ObservationBinding['state'], + provenance: binding.provenance as ObservationBinding['provenance'], + ...(binding.confidence ? { confidence: Number(binding.confidence) } : {}), + ...(binding.matcherVersion ? { matcherVersion: binding.matcherVersion } : {}), + evidence: binding.evidence as MatchEvidence[], + createdAt: binding.createdAt.toISOString(), + updatedAt: binding.updatedAt.toISOString(), + }), + ), + relations: relations.map( + (relation): EventRelation => ({ + id: relation.id, + fromEventId: relation.fromEventId, + toEventId: relation.toEventId, + type: relation.type as EventRelation['type'], + createdAt: relation.createdAt.toISOString(), + }), + ), + } +} + +export const listEconomicEvents = async (db: Db, request: EventsListRequest = {}) => { + const from = asDate(request.from, 'events.list from') + const to = asDate(request.to, 'events.list to') + if (from && to && from > to) throw new Error('events.list from must not be after to') + const limit = request.limit ?? 100 + if (!Number.isSafeInteger(limit) || limit < 1 || limit > 500) { + throw new Error('events.list limit must be an integer between 1 and 500') + } + const rows = await db.select().from(economicEvents).orderBy(asc(economicEvents.occurredFrom)) + const filtered = rows.filter( + (event) => + (!from || event.occurredTo >= from) && + (!to || event.occurredFrom <= to) && + (!request.states || request.states.includes(event.state as EconomicEvent['state'])), + ) + const items = await Promise.all(filtered.slice(0, limit).map((event) => eventView(db, event))) + return { items } +} + +export const getEconomicEvent = async (db: Db, eventId: string) => { + if (!eventId) throw new Error('eventId is required') + const [event] = await db + .select() + .from(economicEvents) + .where(eq(economicEvents.id, eventId)) + .limit(1) + if (!event) throw new Error('economic event not found') + return eventView(db, event) +} + +const asAccountLink = (row: typeof accountLinks.$inferSelect): AccountLink => ({ + id: row.id, + sourceAccountId: row.sourceAccountId, + targetAccountId: row.targetAccountId, + type: row.type as AccountLink['type'], + ...(row.instrument ? { instrument: row.instrument as AccountLink['instrument'] } : {}), + ...(row.validFrom ? { validFrom: row.validFrom.toISOString() } : {}), + ...(row.validTo ? { validTo: row.validTo.toISOString() } : {}), + source: row.source as AccountLink['source'], + confirmed: row.confirmed, +}) + +export const listAccountLinks = async (db: Db) => + (await db.select().from(accountLinks).orderBy(asc(accountLinks.id))).map(asAccountLink) + +export const listFinancialAccounts = async (db: Db): Promise => + (await db.select().from(financialAccounts).orderBy(asc(financialAccounts.createdAt))).map( + (account) => ({ + id: account.id, + profileId: account.profileId, + connectorTypeId: account.connectorTypeId, + institutionId: account.institutionId, + providerAccountId: account.providerAccountId, + kind: account.kind as FinancialAccount['kind'], + createdAt: account.createdAt.toISOString(), + updatedAt: account.updatedAt.toISOString(), + }), + ) + +export const upsertAccountLink = async (db: Db, input: AccountLinkInput): Promise => { + if (!input.sourceAccountId || !input.targetAccountId) { + throw new Error('sourceAccountId and targetAccountId are required') + } + if (input.sourceAccountId === input.targetAccountId) { + throw new Error('an account link must connect two different accounts') + } + const validFrom = asDate(input.validFrom, 'account link validFrom') + const validTo = asDate(input.validTo, 'account link validTo') + if (validFrom && validTo && validFrom > validTo) { + throw new Error('account link validFrom must not be after validTo') + } + const accounts = await db.select({ id: financialAccounts.id }).from(financialAccounts) + const accountIds = new Set(accounts.map((account) => account.id)) + if (!accountIds.has(input.sourceAccountId) || !accountIds.has(input.targetAccountId)) { + throw new Error('account link references an unknown financial account') + } + const id = input.id ?? randomId('account-link') + const [existing] = await db.select().from(accountLinks).where(eq(accountLinks.id, id)).limit(1) + if (existing && existing.source !== 'user') { + throw new Error('provider or inferred account links cannot be modified through this operation') + } + const values = { + id, + sourceAccountId: input.sourceAccountId, + targetAccountId: input.targetAccountId, + type: input.type, + instrument: input.instrument ?? null, + validFrom: validFrom ?? null, + validTo: validTo ?? null, + source: 'user' as const, + confirmed: true, + } + await db + .insert(accountLinks) + .values(values) + .onConflictDoUpdate({ target: accountLinks.id, set: values }) + const [saved] = await db.select().from(accountLinks).where(eq(accountLinks.id, id)).limit(1) + if (!saved) throw new Error('account link was not saved') + return asAccountLink(saved) +} + +export const deleteAccountLink = async (db: Db, id: string) => { + if (!id) throw new Error('id is required') + const [existing] = await db.select().from(accountLinks).where(eq(accountLinks.id, id)).limit(1) + if (!existing) throw new Error('account link not found') + if (existing.source !== 'user') + throw new Error('provider or inferred account links cannot be deleted') + await db.delete(accountLinks).where(eq(accountLinks.id, id)) +} + +const moneyAmount = (amount: Amount | null): Extract | undefined => + amount?.kind === 'money' ? amount : undefined + +const sameAmount = (left: Amount | null, right: Amount | null) => { + const source = moneyAmount(left) + const target = moneyAmount(right) + return Boolean( + source && + target && + source.money.currency === target.money.currency && + source.money.value === target.money.value, + ) +} + +const candidateKey = (sourceObservationId: string, targetObservationId: string) => + `wallet-topup:${[sourceObservationId, targetObservationId].sort().join(':')}` + +const timeDistanceMs = (left: Transaction, right: Transaction) => + Math.abs(new Date(left.occurredAt).getTime() - new Date(right.occurredAt).getTime()) + +const ledgerForFinancialAccount = async (db: Db, financialAccountId: string) => { + const [existing] = await db + .select() + .from(ledgerAccounts) + .where(eq(ledgerAccounts.financialAccountId, financialAccountId)) + .limit(1) + if (existing) return existing.id + const [financialAccount] = await db + .select() + .from(financialAccounts) + .where(eq(financialAccounts.id, financialAccountId)) + .limit(1) + if (!financialAccount) throw new Error('financial account not found') + const id = `ledger_${financialAccount.id}` + const ledgerClass = financialAccount.kind === 'credit-card' ? 'liability' : 'asset' + await db + .insert(ledgerAccounts) + .values({ + id, + class: ledgerClass, + financialAccountId, + name: financialAccount.providerAccountId, + createdAt: new Date(), + }) + .onConflictDoNothing() + return id +} + +interface StoredObservation { + id: string + accountId: string + transaction: Transaction +} + +const storedObservations = async (db: Db, from: Date, to: Date): Promise => { + const [observations, revisions] = await Promise.all([ + db.select().from(transactionObservations), + db.select().from(transactionObservationRevisions), + ]) + const revisionByObservation = new Map( + revisions.map((revision) => [`${revision.observationId}:${revision.revision}`, revision]), + ) + return observations.flatMap((observation) => { + const revision = revisionByObservation.get(`${observation.id}:${observation.currentRevision}`) + const transaction = revision?.normalized as Transaction | undefined + if (!transaction) return [] + const occurredAt = new Date(transaction.occurredAt) + if (!Number.isFinite(occurredAt.getTime()) || occurredAt < from || occurredAt > to) return [] + return [{ id: observation.id, accountId: observation.accountId, transaction }] + }) +} + +const createWalletTopupProposal = async ( + db: Db, + source: StoredObservation, + target: StoredObservation, + link: AccountLink, +) => { + const key = candidateKey(source.id, target.id) + const [rejected] = await db + .select() + .from(reconciliationDecisions) + .where(eq(reconciliationDecisions.candidateKey, key)) + .limit(1) + if (rejected?.decision === 'rejected') return false + const [existing] = await db + .select() + .from(reconciliationProposals) + .where(eq(reconciliationProposals.candidateKey, key)) + .limit(1) + if (existing) return false + const amount = moneyAmount(source.transaction.amount) + if (!amount) return false + const now = new Date() + const eventId = randomId('economic-event') + const sourceLedgerAccountId = await ledgerForFinancialAccount(db, source.accountId) + const targetLedgerAccountId = await ledgerForFinancialAccount(db, target.accountId) + const sourcePostingId = randomId('posting') + const targetPostingId = randomId('posting') + const sourceBindingId = randomId('binding') + const targetBindingId = randomId('binding') + const sourceTime = new Date(source.transaction.occurredAt) + const targetTime = new Date(target.transaction.occurredAt) + await db.insert(economicEvents).values({ + id: eventId, + kind: 'wallet-topup', + state: 'proposed', + completeness: 'complete', + occurredFrom: sourceTime < targetTime ? sourceTime : targetTime, + occurredTo: sourceTime < targetTime ? targetTime : sourceTime, + metadata: link.instrument?.network ? { rail: link.instrument.network } : null, + createdAt: now, + updatedAt: now, + }) + await db.insert(eventPostings).values([ + { + id: sourcePostingId, + eventId, + ledgerAccountId: sourceLedgerAccountId, + side: 'credit', + amount, + role: 'source', + }, + { + id: targetPostingId, + eventId, + ledgerAccountId: targetLedgerAccountId, + side: 'debit', + amount, + role: 'destination', + }, + ]) + const evidence: MatchEvidence[] = [ + { kind: 'same-amount', value: amount.money.value, currency: amount.money.currency }, + { kind: 'time-distance', milliseconds: timeDistanceMs(source.transaction, target.transaction) }, + { kind: 'account-link', accountLinkId: link.id }, + { kind: 'rule', ruleId: 'wallet-topup-v1' }, + ] + await db.insert(observationBindings).values([ + { + id: sourceBindingId, + observationId: source.id, + eventId, + postingIds: [sourcePostingId], + state: 'proposed', + provenance: 'deterministic-rule', + confidence: '0.95', + matcherVersion: 'wallet-topup-v1', + evidence, + createdAt: now, + updatedAt: now, + }, + { + id: targetBindingId, + observationId: target.id, + eventId, + postingIds: [targetPostingId], + state: 'proposed', + provenance: 'deterministic-rule', + confidence: '0.95', + matcherVersion: 'wallet-topup-v1', + evidence, + createdAt: now, + updatedAt: now, + }, + ]) + await db.insert(reconciliationProposals).values({ + id: randomId('reconciliation-proposal'), + candidateKey: key, + eventId, + score: '0.95', + state: 'proposed', + createdAt: now, + }) + return true +} + +/** Generates only unambiguous debit-funded wallet top-up proposals. */ +export const reconcileRange = async (db: Db, from: Date, to: Date) => { + const [links, observations] = await Promise.all([ + listAccountLinks(db), + storedObservations(db, from, to), + ]) + let created = 0 + for (const link of links.filter( + (candidate) => candidate.confirmed && candidate.type === 'funds', + )) { + const source = observations.filter( + (observation) => + observation.accountId === link.sourceAccountId && + observation.transaction.direction === 'debit', + ) + const target = observations.filter( + (observation) => + observation.accountId === link.targetAccountId && + observation.transaction.direction === 'credit' && + observation.transaction.kind === 'charge', + ) + const candidates = source.flatMap((sourceObservation) => + target + .filter( + (targetObservation) => + sameAmount( + sourceObservation.transaction.amount, + targetObservation.transaction.amount, + ) && + timeDistanceMs(sourceObservation.transaction, targetObservation.transaction) <= + 36 * 60 * 60_000, + ) + .map((targetObservation) => ({ sourceObservation, targetObservation })), + ) + for (const candidate of candidates) { + const sourceCount = candidates.filter( + (other) => other.sourceObservation.id === candidate.sourceObservation.id, + ).length + const targetCount = candidates.filter( + (other) => other.targetObservation.id === candidate.targetObservation.id, + ).length + if (sourceCount !== 1 || targetCount !== 1) continue + if ( + await createWalletTopupProposal( + db, + candidate.sourceObservation, + candidate.targetObservation, + link, + ) + ) { + created += 1 + } + } + } + return created +} + +export const enqueueReconciliation = async (db: Db, from: Date, to: Date) => { + if (from > to) throw new Error('reconciliation job from must not be after to') + await db.insert(reconciliationJobs).values({ + id: randomId('reconciliation-job'), + from, + to, + state: 'queued', + createdAt: new Date(), + }) +} + +export const runQueuedReconciliation = async (db: Db) => { + const [job] = await db + .select() + .from(reconciliationJobs) + .where(eq(reconciliationJobs.state, 'queued')) + .orderBy(asc(reconciliationJobs.createdAt)) + .limit(1) + if (!job) return 0 + const startedAt = new Date() + await db + .update(reconciliationJobs) + .set({ state: 'running', startedAt, error: null }) + .where(eq(reconciliationJobs.id, job.id)) + try { + const created = await reconcileRange(db, job.from, job.to) + await db + .update(reconciliationJobs) + .set({ state: 'completed', completedAt: new Date() }) + .where(eq(reconciliationJobs.id, job.id)) + return created + } catch (cause) { + const message = cause instanceof Error ? cause.message : String(cause) + await db + .update(reconciliationJobs) + .set({ state: 'failed', completedAt: new Date(), error: message }) + .where(eq(reconciliationJobs.id, job.id)) + throw cause + } +} + +const proposalObservations = async (db: Db, eventId: string) => { + const bindings = await db + .select() + .from(observationBindings) + .where(eq(observationBindings.eventId, eventId)) + const revisions = await db.select().from(transactionObservationRevisions) + const observations = await db.select().from(transactionObservations) + const revisionByObservation = new Map( + revisions.map((revision) => [ + `${revision.observationId}:${revision.revision}`, + revision.normalized, + ]), + ) + const currentRevision = new Map( + observations.map((observation) => [observation.id, observation.currentRevision]), + ) + return bindings.flatMap((binding) => { + const revision = currentRevision.get(binding.observationId) + const transaction = + revision && revisionByObservation.get(`${binding.observationId}:${revision}`) + return transaction ? [transaction as Transaction] : [] + }) +} + +export const listReconciliationProposals = async ( + db: Db, + request: ReconciliationProposalRequest = {}, +) => { + const limit = request.limit ?? 100 + const rows = await db + .select() + .from(reconciliationProposals) + .orderBy(asc(reconciliationProposals.createdAt)) + const filtered = rows.filter( + (proposal) => !request.states || request.states.includes(proposal.state as never), + ) + const items = await Promise.all( + filtered.slice(0, limit).map(async (proposal): Promise => { + const view = await getEconomicEvent(db, proposal.eventId) + return { + id: proposal.id, + candidateKey: proposal.candidateKey, + event: view.event, + observations: await proposalObservations(db, proposal.eventId), + bindings: view.bindings, + score: Number(proposal.score), + createdAt: proposal.createdAt.toISOString(), + } + }), + ) + return { items } +} + +export const confirmReconciliationProposal = async (db: Db, proposalId: string) => { + const [proposal] = await db + .select() + .from(reconciliationProposals) + .where(eq(reconciliationProposals.id, proposalId)) + .limit(1) + if (!proposal) throw new Error('reconciliation proposal not found') + if (proposal.state !== 'proposed') throw new Error('reconciliation proposal is no longer pending') + const now = new Date() + const bindings = await db + .select() + .from(observationBindings) + .where(eq(observationBindings.eventId, proposal.eventId)) + await db.transaction(async (tx) => { + await tx + .update(economicEvents) + .set({ state: 'confirmed', updatedAt: now }) + .where(eq(economicEvents.id, proposal.eventId)) + for (const binding of bindings) { + await tx + .update(observationBindings) + .set({ state: 'confirmed', provenance: 'user', updatedAt: now }) + .where(eq(observationBindings.id, binding.id)) + } + await tx + .update(reconciliationProposals) + .set({ state: 'confirmed' }) + .where(eq(reconciliationProposals.id, proposal.id)) + await tx.insert(reconciliationDecisions).values({ + id: randomId('reconciliation-decision'), + proposalId: proposal.id, + candidateKey: proposal.candidateKey, + decision: 'confirmed', + decidedAt: now, + }) + }) + return getEconomicEvent(db, proposal.eventId) +} + +export const rejectReconciliationProposal = async (db: Db, proposalId: string, reason?: string) => { + const [proposal] = await db + .select() + .from(reconciliationProposals) + .where(eq(reconciliationProposals.id, proposalId)) + .limit(1) + if (!proposal) throw new Error('reconciliation proposal not found') + if (proposal.state !== 'proposed') throw new Error('reconciliation proposal is no longer pending') + const now = new Date() + await db.transaction(async (tx) => { + await tx + .update(economicEvents) + .set({ state: 'superseded', updatedAt: now }) + .where(eq(economicEvents.id, proposal.eventId)) + await tx + .update(observationBindings) + .set({ state: 'rejected', updatedAt: now }) + .where(eq(observationBindings.eventId, proposal.eventId)) + await tx + .update(reconciliationProposals) + .set({ state: 'rejected' }) + .where(eq(reconciliationProposals.id, proposal.id)) + await tx.insert(reconciliationDecisions).values({ + id: randomId('reconciliation-decision'), + proposalId: proposal.id, + candidateKey: proposal.candidateKey, + decision: 'rejected', + ...(reason ? { reason } : {}), + decidedAt: now, + }) + }) +} diff --git a/apps/mnie-server/src/rpc/handler.ts b/apps/mnie-server/src/rpc/handler.ts index 56333f8..0a6a1fa 100644 --- a/apps/mnie-server/src/rpc/handler.ts +++ b/apps/mnie-server/src/rpc/handler.ts @@ -45,6 +45,23 @@ const assertOperationScope = (state: RpcSocketState, operation: string, input?: if (!(state.scopes ?? []).includes(required)) throw new Error(`missing OAuth scope: ${required}`) } +const isReconciliationWrite = (operation: string) => + operation === 'reconciliation.confirm' || + operation === 'reconciliation.reject' || + operation === 'account-links.upsert' || + operation === 'account-links.delete' + +const assertWorkspaceOperationScope = (state: RpcSocketState, operation: string) => { + assertReadScope(state) + if ( + state.apiKeyId && + isReconciliationWrite(operation) && + !(state.scopes ?? []).includes('reconcile') + ) { + throw new Error('missing OAuth scope: reconcile') + } +} + export const tradeLimitParams = ( input: unknown, transactionAmount?: { currency: string; value: string }, @@ -105,12 +122,12 @@ export const handleRpc = async ( } if (request.method === 'workspace.invoke') { - assertReadScope(state) const params = objectParams(request.params) const operation = String(params.operation ?? '') if (!WORKSPACE_OPERATIONS.includes(operation as (typeof WORKSPACE_OPERATIONS)[number])) { return rpcError(request.id, -32601, 'workspace operation not found') } + assertWorkspaceOperationScope(state, operation) return rpcResult( request.id, await invokeWorkspace(db, providers, operation, objectParams(params.input)), diff --git a/apps/mnie-server/src/rpc/workspace.ts b/apps/mnie-server/src/rpc/workspace.ts index d530b5d..345b728 100644 --- a/apps/mnie-server/src/rpc/workspace.ts +++ b/apps/mnie-server/src/rpc/workspace.ts @@ -1,15 +1,35 @@ -import type { HistoryListRequest } from '@mnie/types' +import type { EventsListRequest, HistoryListRequest } from '@mnie/types' import { fetchAssetValuation } from '../assets' import type { Db } from '../db' import { listHistory } from '../history' import { loadPortfolioOverview, type OverviewProfile } from '../portfolio-overview' import type { ProviderRegistry } from '../providers/registry' +import { + confirmReconciliationProposal, + deleteAccountLink, + getEconomicEvent, + listAccountLinks, + listFinancialAccounts, + listEconomicEvents, + listReconciliationProposals, + rejectReconciliationProposal, + upsertAccountLink, +} from '../reconciliation' export const WORKSPACE_OPERATIONS = [ 'profiles.list', 'portfolio.valuation.get', 'portfolio.overview.get', 'history.list', + 'financial-accounts.list', + 'events.list', + 'events.get', + 'reconciliation.proposals.list', + 'reconciliation.confirm', + 'reconciliation.reject', + 'account-links.list', + 'account-links.upsert', + 'account-links.delete', ] as const export const invokeWorkspace = async ( @@ -38,6 +58,28 @@ export const invokeWorkspace = async ( return listHistory(db, providers, input as HistoryListRequest & { profileIds?: string[] }) } + if (operation === 'events.list') return listEconomicEvents(db, input as EventsListRequest) + if (operation === 'financial-accounts.list') return listFinancialAccounts(db) + if (operation === 'events.get') return getEconomicEvent(db, String(input.eventId ?? '')) + if (operation === 'reconciliation.proposals.list') return listReconciliationProposals(db, input) + if (operation === 'reconciliation.confirm') { + return confirmReconciliationProposal(db, String(input.proposalId ?? '')) + } + if (operation === 'reconciliation.reject') { + await rejectReconciliationProposal( + db, + String(input.proposalId ?? ''), + typeof input.reason === 'string' ? input.reason : undefined, + ) + return undefined + } + if (operation === 'account-links.list') return listAccountLinks(db) + if (operation === 'account-links.upsert') return upsertAccountLink(db, input as never) + if (operation === 'account-links.delete') { + await deleteAccountLink(db, String(input.id ?? '')) + return undefined + } + if (operation === 'portfolio.valuation.get') { const baseCurrency = String(input.baseCurrency ?? 'JPY') const requested = Array.isArray(input.profileIds) diff --git a/apps/mnie-server/src/security/oauth-provider.ts b/apps/mnie-server/src/security/oauth-provider.ts index 0b16d48..812b822 100644 --- a/apps/mnie-server/src/security/oauth-provider.ts +++ b/apps/mnie-server/src/security/oauth-provider.ts @@ -16,7 +16,7 @@ import { randomToken, sha256 } from './crypto' const CODE_TTL_MS = 10 * 60 * 1000 const ACCESS_TOKEN_TTL_SECONDS = 60 * 60 const REFRESH_TOKEN_TTL_MS = 30 * 24 * 60 * 60 * 1000 -const DEFAULT_SCOPES = ['read', 'trade'] as const +const DEFAULT_SCOPES = ['read', 'trade', 'reconcile'] as const export const createOAuthAuthorizationCode = async ( db: Db, diff --git a/apps/mnie-ui/src/App.vue b/apps/mnie-ui/src/App.vue index 6969fbe..8841eca 100644 --- a/apps/mnie-ui/src/App.vue +++ b/apps/mnie-ui/src/App.vue @@ -13,6 +13,7 @@ import OAuthApprovalPanel from './features/oauth/OAuthApprovalPanel.vue' import { useOAuthApproval } from './features/oauth/useOAuthApproval' import OrderDialogs from './features/orders/OrderDialogs.vue' import PortfolioView from './features/portfolio/PortfolioView.vue' +import ReconciliationView from './features/reconciliation/ReconciliationView.vue' import SettingsView from './features/settings/SettingsView.vue' import TradeView from './features/trade/TradeView.vue' import { useTradingSession } from './features/trading/useTradingSession' @@ -461,6 +462,8 @@ onUnmounted(() => { @confirm-order="askPlaceOrder" /> + + + bindings: Array<{ id: string; evidence: Array<{ kind: string }> }> +} + +export interface FinancialAccount { + id: string + connectorTypeId: string + providerAccountId: string + kind: string +} + export type ProfileAvailability = | { ok: true; checkedAt?: string; operations?: Record } | { @@ -408,6 +436,23 @@ export const listHistory = ( } = {}, ) => workspaceRequest<{ items: HistoryItem[]; errors: HistoryListError[] }>('history.list', input) +export const listReconciliationProposals = () => + workspaceRequest<{ items: ReconciliationProposal[] }>('reconciliation.proposals.list') + +export const listFinancialAccounts = () => + workspaceRequest('financial-accounts.list') + +export const listAccountLinks = () => workspaceRequest('account-links.list') + +export const saveAccountLink = (input: Record) => + workspaceRequest('account-links.upsert', input) + +export const confirmReconciliationProposal = (proposalId: string) => + workspaceRequest('reconciliation.confirm', { proposalId }) + +export const rejectReconciliationProposal = (proposalId: string, reason?: string) => + workspaceRequest('reconciliation.reject', { proposalId, ...(reason ? { reason } : {}) }) + export const checkAccountProfileAvailability = () => adminRequest<{ availability: Record }>( 'profiles.availability.cached', diff --git a/apps/mnie-ui/src/components/ApiKeyPolicyEditor.vue b/apps/mnie-ui/src/components/ApiKeyPolicyEditor.vue index 651f3ae..ab17e8f 100644 --- a/apps/mnie-ui/src/components/ApiKeyPolicyEditor.vue +++ b/apps/mnie-ui/src/components/ApiKeyPolicyEditor.vue @@ -34,7 +34,7 @@ const tradingMethods = [ const tradingMethodSet = new Set(tradingMethods) const readMethods = rpcMethods.filter((method) => !tradingMethodSet.has(method)) -const scopes = ['read', 'trade'] as const +const scopes = ['read', 'trade', 'reconcile'] as const const toggleScope = (scope: string) => { const current = settings.value.scopes ?? [] diff --git a/apps/mnie-ui/src/constants/provider.ts b/apps/mnie-ui/src/constants/provider.ts index 0765c3d..bf80dea 100644 --- a/apps/mnie-ui/src/constants/provider.ts +++ b/apps/mnie-ui/src/constants/provider.ts @@ -2,7 +2,7 @@ import type { AccountProfile } from '../api' export const defaultProviderColors: Record = { 'smbc-direct': '#005b47', - mobilesuica: '#2F8E3C', + 'mobile-suica': '#2F8E3C', 'paypay-bank': '#f5bac4', 'paypay-sec': '#ff003c', sbisec: '#0a3e86', diff --git a/apps/mnie-ui/src/features/reconciliation/ReconciliationView.vue b/apps/mnie-ui/src/features/reconciliation/ReconciliationView.vue new file mode 100644 index 0000000..0b5da0b --- /dev/null +++ b/apps/mnie-ui/src/features/reconciliation/ReconciliationView.vue @@ -0,0 +1,169 @@ + + + diff --git a/apps/mnie-ui/src/features/settings/SettingsView.vue b/apps/mnie-ui/src/features/settings/SettingsView.vue index 3783ef7..83224fb 100644 --- a/apps/mnie-ui/src/features/settings/SettingsView.vue +++ b/apps/mnie-ui/src/features/settings/SettingsView.vue @@ -557,7 +557,7 @@ const saveEditingApiKey = () => { {
@@ -595,7 +589,7 @@ const saveEditingApiKey = () => {

{{ job.label }}

{{ - selectedProvider === 'mobilesuica' + selectedProvider === 'mobile-suica' ? '5 分ごとに利用履歴を取得して接続を維持します。' : '5 分ごとに接続を維持します。' }} @@ -756,7 +750,7 @@ const saveEditingApiKey = () => {

-