Files
fusion/packages/core/src/gitlab-issue-analytics.ts
gsxdsm c15c78feeb feat: migrate storage from SQLite to PostgreSQL (#1793)
# Migrate storage from SQLite to PostgreSQL — full dashboard cutover

Migrates Fusion's storage layer to the embedded PostgreSQL
`AsyncDataLayer` (the default backend) and **completes the
satellite-store + feature cutover** so every dashboard and Command
Center surface works in PG mode.

## Status — every surface works in embedded-PG mode

Verified live against a running embedded-Postgres dashboard (all
**200**, zero 5xx) and gate-tested (**23 files / 99 tests** on embedded
PG, plus engine-core 294 and ci-shape 63 in the blocking merge gate;
core/engine/cli/dashboard typecheck clean).

| Area | Surfaces | State |
|---|---|---|
| Satellite stores | workflows, todos, insights, research, missions,
goals, mailbox | ✅ |
| Views | artifacts, documents, evals | ✅ |
| Command Center | activity, productivity, team, tokens, tools,
**workflows**, **github**, **signals**, **plugin-activations**, **live**
(all 10) | ✅ |
| Run execution | insight generation, research run execution | ✅
(store-path; AI step needs a provider) |
| Live updates | SSE push for mission/research/insight events | ✅ |
| Workflow editing | create / update / delete / select (+ id counter) |
✅ |
| Engine | mission autopilot, incident-signal ingestion, regression
storm-guard, agent wake-on-message | ✅ |
| Core | tasks, agents, secrets, automations, memory, chat, usage, PRs,
git | ✅ |

## Approach

Each satellite store gets an `Async<Store>` wrapper exposing the sync
store's method names over the existing `async-*-store.ts` helpers;
`get<Store>Store()` returns a `Sync | Async` union; consumers `await`
(harmless on sync), and engine/CLI paths that can't convert use
`instanceof Sync` graceful fallback. Analytics aggregators branch on
`"ping" in dbOrLayer` to run schema-qualified raw SQL over `project.*`
(snake_case) in PG. Executors/orchestrators/autopilot are
await-converted to drive the union store; the async store wrappers
extend `EventEmitter` so SSE live-push fires in both backends.

Not-yet-ported capabilities degrade gracefully (never 500) and are
individually called out in commits.

## Sync with main

The branch is kept continuously merged with `main` (currently through
FN-7845, 2026-07-12); the earlier "final rebase deferred" note no longer
applies. Use **Create a merge commit** (or squash) to land it — GitHub's
rebase-merge cannot replay a merge-maintained branch.

## Residual Review Findings

Multi-agent code review of the PostgreSQL satellite-store ports (U1–U5)
applied 3 safe fixes (see `fix(review): apply autofix feedback`). The
following are **real but gated** — recorded here as follow-up work
rather than auto-applied. All are SQLite→PostgreSQL
**concurrency/atomicity regressions**: the sync stores were immune only
by SQLite's single-writer, single-threaded-handler execution; the async
ports open multi-await read-modify-write windows. **Reachability is low
today** because the execution engines that generate concurrent same-run
mutations (insight run executor, research orchestrator/dispatcher) are
`instanceof`-gated to sync mode in PG. No process-crash class survived
(all engine fallbacks correctly guard the sync store).

- **[P1] Research `appendResearchEvent` dual-write is non-atomic**
(`packages/core/src/async-research-store.ts`, corroborated: adversarial
+ reliability). The `research_run_events` insert (own transaction) and
the `run.events` jsonb update are separate writes — a crash between
them, or two concurrent appends, splits the table count from the jsonb
array. **Fix:** perform the seq-insert and the jsonb update in one
`layer.transactionImmediate`.
- **[P1] Research run terminal-reversion via stale full-row persist**
(`async-research-store.ts` `persistResearchRun`/`updateResearchStatus`).
Concurrent `PATCH /runs/:id/status` + `POST /runs/:id/events` can revert
a terminal run to `running` by overwriting the whole row, bypassing the
transition guard. **Fix:** scoped column `UPDATE`s with a `WHERE status
…` guard, or optimistic version column.
- **[P2] `updateResearchRun`/`updateInsightRun` read-then-write TOCTOU**
— concurrent PATCHes last-writer-wins on the lifecycle merge. **Fix:**
`SELECT … FOR UPDATE` / enclosing transaction.
- **[P2] `upsertRun`/`createRunOrThrowConflict` check-then-create race**
(`async-insight-store.ts`) — two callers can each create an "active"
run. **Fix:** partial unique index on `(projectId, trigger) WHERE status
IN ('pending','running')`.
- **[P3] `createResearchRetryRun` return-value divergence** — sync
returns the pre-update `queued` snapshot; async returns the reloaded
`retry_waiting` run (persisted state is identical). Pick one side for
cross-backend parity.
- **[P2/perf] Mission `getMissionWithHierarchy`/`getMissionHealth` N+1
fan-out** — O(milestones×slices) sequential round-trips hold one pool
slot per request; can starve the pool for large hierarchies. **Fix:**
batched/joined reads.
- **Testing gaps:** no PG-mode concurrency tests (interleaved
status/event mutations), no sync↔async parity assertion for the
lifecycle-error codes, and no mission status/health rollup parity test
vs the sync `MissionStore`.

~~Out of scope (deferred): AI run *execution* (insight/research) +
mission autopilot + live SSE mission events remain sync-gated/degraded
in PG mode.~~ **Since ported** — insight/research run execution, mission
autopilot, and SSE live push all run on the async layer now, which also
makes the concurrency findings above genuinely reachable; they remain
open follow-ups.







---

## Update — 2026-07-12: production-readiness hardening & live acceptance

Everything below landed on this branch since the description above was
written:

**Production blockers from review — fixed**
- `recoverStaleTransitionPending` ported to the async layer (backend
moves write + clear the crash-safe marker; startup/maintenance sweeps no
longer throw).
- Lost-update class fixed: `atomicWriteTaskJson`/`WithAudit` write
changed columns only (full-row upserts silently resurrected stale fields
across concurrent store instances — the "task stuck unplanned forever"
bug).
- First-boot **auto-migration**: booting the PG backend over a project
with a legacy `fusion.db` migrates it automatically (loud failure,
SQLite kept as backup), and the dashboard shows a one-time **"your data
was migrated" banner** with the backup paths and a Need-help Discord
link.
- `pg_dump`/`pg_restore` discovered from common install locations for
embedded-mode backups.
- The PG suite is part of the blocking merge gate (`test:pg-gate`).

**Multi-project isolation (PR #2007, merged into this branch)**
- `project_id` partition key on tasks / archived tasks / config,
`taskProjectScope` threaded through every scan/claim/count, per-project
config rows, layer bound to the project at startup.
- Review P1 follow-up: the shared cold-storage `archive.archived_tasks`
table is also partitioned and all archived-board reads/counts/searches
are scoped.
- Schema drift self-heal generalized to schema-qualified columns so
existing databases upgrade in place.

**Other changes**
- Node settings sync **removed** in PG mode (409
`settings-sync-disabled-postgres`) — nodes share state by connecting to
the same database; auth sync kept (per-machine file).
- Perf (review findings): `listTasks` pushes column filter + ORDER BY +
LIMIT/OFFSET into SQL; `getConversation` capped to the most recent 200
messages.
- Fixed a false "operator action required" pause-abort log fired on
every successfully auto-merged task.

**Live acceptance — PASSED (2026-07-12)**
A sandboxed instance (isolated HOME, embedded PG, real Opus executor)
ran a task through the complete cycle: create → triage (AI spec) →
execute → in-review → AI squash-merge landed on the project's `main` →
done. A write+read sweep of every data surface (settings, comments,
documents, attachments + artifact bridge + artifact edit, chat with real
generation, goals, missions, agent mail, secrets, workflows, memory, CC
analytics) was green on embedded PG.

**Known remaining work**
- The per-project `config` PK re-key has no upgrade path for
pre-isolation embedded-PG databases (needs a real `DROP
CONSTRAINT`/re-key migration; fresh databases are fine).
- `pg_dump`/`pg_restore` binaries are not yet bundled in release
artifacts (PATH/common-location discovery only).
- The satellite-store concurrency findings listed above.

---------

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Co-authored-by: Phil Larson <hello@phillarson.xyz>
Co-authored-by: fusion-merge <fusion-merge@local>
2026-07-13 19:07:58 -07:00

311 lines
11 KiB
TypeScript

import { sql } from "drizzle-orm";
import type { Database } from "./db.js";
import type { AsyncDataLayer } from "./postgres/data-layer.js";
/**
* FNXC:CommandCenterGitLab 2026-07-02-00:00:
* GitLab analytics must be provider-isolated while sharing the generic sourceIssue storage columns with GitHub. Filed counts read only `gitlabTracking.item`; fixed counts read only `sourceIssueProvider = "gitlab"`, using exact `sourceIssueClosedAt` when present and `updatedAt` only as an approximation fallback.
*/
export interface GitlabIssueAnalyticsQuery {
/** ISO-8601 lower bound (inclusive). */
from?: string;
/** ISO-8601 upper bound (inclusive). */
to?: string;
}
export interface GitlabIssueDailyPoint {
/** UTC date, `YYYY-MM-DD`. */
date: string;
/** Fusion-created GitLab items filed on this date. */
filed: number;
/** Imported GitLab issue/MR tasks completed on this date. */
fixed: number;
}
export interface GitlabIssueProjectBreakdown {
/** GitLab project/group key; `(unknown)` when historical data lacks it. */
project: string;
filed: number;
fixed: number;
}
export interface GitlabResolvedIssue {
/** Fusion task that resolved the imported GitLab source item. */
taskId: string;
/** Fusion task title at aggregation time. */
taskTitle: string;
/** GitLab project/group key; `(unknown)` when historical data lacks it. */
project: string;
/** GitLab issue or merge request IID when stored. */
issueNumber: number | null;
/** Source GitLab item URL when available. */
url: string | null;
/** ISO timestamp used for range filtering and ordering. */
resolvedAt: string;
/** True when `sourceIssueClosedAt` supplied `resolvedAt`; false when `updatedAt` was used. */
resolvedAtExact: boolean;
}
export interface GitlabIssueAnalytics {
from: string | null;
to: string | null;
/** Fusion-created GitLab tracked items in range. Undated tracked items are included because no date can be honestly inferred. */
filed: number;
/** Imported GitLab source tasks currently in `done`, filtered by exact `sourceIssueClosedAt` when present with `updatedAt` fallback. */
fixed: number;
/** Filed minus fixed. */
net: number;
/** Filed/fixed counts grouped by UTC day, ascending. */
daily: GitlabIssueDailyPoint[];
/** Filed/fixed counts grouped by GitLab project/group, descending by total activity. */
byProject: GitlabIssueProjectBreakdown[];
/** Imported GitLab source items completed in range, most-recently resolved first. */
resolved: GitlabResolvedIssue[];
}
interface GitlabTrackingRow {
gitlabTracking: string | null;
}
interface FixedIssueRow {
id: string;
title: string | null;
sourceIssueRepository: string | null;
sourceIssueNumber: number | null;
sourceIssueUrl: string | null;
sourceIssueClosedAt: string | null;
updatedAt: string | null;
}
interface GitlabTrackedItemLike {
iid?: unknown;
projectPath?: unknown;
groupPath?: unknown;
projectId?: unknown;
createdAt?: unknown;
}
interface GitlabTrackingLike {
item?: GitlabTrackedItemLike;
}
function isInRange(iso: string, query: GitlabIssueAnalyticsQuery): boolean {
const t = Date.parse(iso);
if (!Number.isFinite(t)) return false;
if (query.from !== undefined && t < Date.parse(query.from)) return false;
if (query.to !== undefined && t > Date.parse(query.to)) return false;
return true;
}
function dayKey(iso: string): string | null {
const t = Date.parse(iso);
if (!Number.isFinite(t)) return null;
return new Date(t).toISOString().slice(0, 10);
}
function projectFromItem(item: GitlabTrackedItemLike): string {
const projectPath = typeof item.projectPath === "string" ? item.projectPath.trim() : "";
if (projectPath) return projectPath;
const groupPath = typeof item.groupPath === "string" ? item.groupPath.trim() : "";
if (groupPath) return groupPath;
if (typeof item.projectId === "number" && Number.isFinite(item.projectId)) return String(item.projectId);
return "(unknown)";
}
function addDaily(
daily: Map<string, { filed: number; fixed: number }>,
date: string,
kind: "filed" | "fixed",
): void {
const current = daily.get(date) ?? { filed: 0, fixed: 0 };
current[kind] += 1;
daily.set(date, current);
}
function addProject(
byProject: Map<string, { filed: number; fixed: number }>,
project: string,
kind: "filed" | "fixed",
): void {
const current = byProject.get(project) ?? { filed: 0, fixed: 0 };
current[kind] += 1;
byProject.set(project, current);
}
/**
* Aggregate locally persisted GitLab issue and merge-request analytics for the Command Center.
* Empty ranges return zeroed structures, never null collections. Bounds are
* inclusive. Malformed historical `gitlabTracking` JSON is ignored rather than
* failing the entire analytics request.
*
* FNXC:PostgresCutover 2026-07-04-00:00:
* Now accepts a `Database | AsyncDataLayer` and is async. In backend
* (PostgreSQL) mode it branches on `"ping" in dbOrLayer` and reads the real
* `project.tasks` rows (gitlab_tracking is jsonb — already parsed — and the
* source_issue_* columns are snake_case); the sync SQLite branch is unchanged.
* Mirrors aggregateGithubIssueAnalytics.
*/
export async function aggregateGitlabIssueAnalytics(
dbOrLayer: Database | AsyncDataLayer,
query: GitlabIssueAnalyticsQuery = {},
): Promise<GitlabIssueAnalytics> {
if ("ping" in dbOrLayer) {
return aggregateGitlabIssueAnalyticsAsync(dbOrLayer, query);
}
const db = dbOrLayer as Database;
const filedRows = db
.prepare(
"SELECT gitlabTracking FROM tasks WHERE gitlabTracking IS NOT NULL AND gitlabTracking NOT IN ('', '{}')",
)
.all() as GitlabTrackingRow[];
// Sync rows store gitlabTracking as a JSON string; parse (skip malformed).
const filedTrackings: GitlabTrackingLike[] = [];
for (const row of filedRows) {
if (!row.gitlabTracking) continue;
let parsed: unknown;
try {
parsed = JSON.parse(row.gitlabTracking);
} catch {
continue;
}
filedTrackings.push(parsed as GitlabTrackingLike);
}
const fixedRows = db
.prepare(
`SELECT id, title, sourceIssueRepository, sourceIssueNumber, sourceIssueUrl, sourceIssueClosedAt, updatedAt FROM tasks WHERE sourceIssueProvider = 'gitlab' AND "column" = 'done'`,
)
.all() as FixedIssueRow[];
return buildGitlabIssueAnalytics(filedTrackings, fixedRows, query);
}
/**
* FNXC:PostgresCutover 2026-07-04-00:00:
* PostgreSQL fetch path for {@link aggregateGitlabIssueAnalytics}. gitlab_tracking
* is jsonb (postgres-js returns it already parsed, so no JSON.parse), and the
* `gitlab_tracking::text <> '{}'` predicate mirrors the sync `NOT IN ('', '{}')`
* empty-object skip. Fixed-issue columns are aliased back to their camelCase
* row shape; source_issue_number coerces to number|null.
*/
async function aggregateGitlabIssueAnalyticsAsync(
layer: AsyncDataLayer,
query: GitlabIssueAnalyticsQuery,
): Promise<GitlabIssueAnalytics> {
const filedRaw = (await layer.db.execute(
sql`SELECT gitlab_tracking AS "gitlabTracking" FROM project.tasks
WHERE gitlab_tracking IS NOT NULL AND gitlab_tracking::text <> '{}'`,
)) as Array<{ gitlabTracking: unknown }>;
const filedTrackings: GitlabTrackingLike[] = [];
for (const row of filedRaw) {
if (row.gitlabTracking == null) continue;
filedTrackings.push(row.gitlabTracking as GitlabTrackingLike);
}
const fixedRaw = (await layer.db.execute(
sql`SELECT
id,
title,
source_issue_repository AS "sourceIssueRepository",
source_issue_number AS "sourceIssueNumber",
source_issue_url AS "sourceIssueUrl",
source_issue_closed_at AS "sourceIssueClosedAt",
updated_at AS "updatedAt"
FROM project.tasks
WHERE source_issue_provider = 'gitlab' AND "column" = 'done'`,
)) as Array<Record<string, unknown>>;
const fixedRows: FixedIssueRow[] = fixedRaw.map((r) => ({
id: String(r.id),
title: (r.title as string | null) ?? null,
sourceIssueRepository: (r.sourceIssueRepository as string | null) ?? null,
sourceIssueNumber: r.sourceIssueNumber == null ? null : Number(r.sourceIssueNumber),
sourceIssueUrl: (r.sourceIssueUrl as string | null) ?? null,
sourceIssueClosedAt: (r.sourceIssueClosedAt as string | null) ?? null,
updatedAt: (r.updatedAt as string | null) ?? null,
}));
return buildGitlabIssueAnalytics(filedTrackings, fixedRows, query);
}
/**
* FNXC:PostgresCutover 2026-07-04-00:00:
* Pure GitLab-issue aggregation shared by the sync (SQLite) and async
* (PostgreSQL) fetch paths. Takes already-parsed `gitlabTracking` objects and
* the fixed-issue rows so both backends produce identical filed/fixed/daily/
* byProject/resolved shapes.
*/
function buildGitlabIssueAnalytics(
filedTrackings: GitlabTrackingLike[],
fixedRows: FixedIssueRow[],
query: GitlabIssueAnalyticsQuery,
): GitlabIssueAnalytics {
const daily = new Map<string, { filed: number; fixed: number }>();
const byProject = new Map<string, { filed: number; fixed: number }>();
let filed = 0;
for (const tracking of filedTrackings) {
const item = tracking.item;
if (!item || typeof item.iid !== "number" || !Number.isFinite(item.iid)) continue;
const createdAt = typeof item.createdAt === "string" ? item.createdAt : undefined;
const hasUsableDate = createdAt !== undefined && dayKey(createdAt) !== null;
if (hasUsableDate && !isInRange(createdAt, query)) continue;
filed += 1;
const project = projectFromItem(item);
addProject(byProject, project, "filed");
if (hasUsableDate && createdAt !== undefined) {
const day = dayKey(createdAt);
if (day !== null) addDaily(daily, day, "filed");
}
}
let fixed = 0;
const resolved: GitlabResolvedIssue[] = [];
for (const row of fixedRows) {
const hasExactResolvedAt = row.sourceIssueClosedAt !== null;
const fixedDate = row.sourceIssueClosedAt ?? row.updatedAt;
if (fixedDate === null || !isInRange(fixedDate, query)) continue;
fixed += 1;
const project = row.sourceIssueRepository?.trim() || "(unknown)";
addProject(byProject, project, "fixed");
const day = dayKey(fixedDate);
if (day !== null) addDaily(daily, day, "fixed");
resolved.push({
taskId: row.id,
taskTitle: row.title ?? "",
project,
issueNumber: typeof row.sourceIssueNumber === "number" ? row.sourceIssueNumber : null,
url: row.sourceIssueUrl?.trim() || null,
resolvedAt: fixedDate,
resolvedAtExact: hasExactResolvedAt,
});
}
resolved.sort((a, b) => {
const byDate = Date.parse(b.resolvedAt) - Date.parse(a.resolvedAt);
return byDate !== 0 ? byDate : a.taskId.localeCompare(b.taskId);
});
return {
from: query.from ?? null,
to: query.to ?? null,
filed,
fixed,
net: filed - fixed,
daily: [...daily.entries()]
.map(([date, counts]) => ({ date, filed: counts.filed, fixed: counts.fixed }))
.sort((a, b) => a.date.localeCompare(b.date)),
byProject: [...byProject.entries()]
.map(([project, counts]) => ({ project, filed: counts.filed, fixed: counts.fixed }))
.sort((a, b) => {
const total = b.filed + b.fixed - (a.filed + a.fixed);
return total !== 0 ? total : a.project.localeCompare(b.project);
}),
resolved,
};
}