apunkt/ktx
1
0
Fork 0
mirror of https://github.com/Kaelio/ktx.git synced 2026-06-19 08:28:06 +02:00
Code Issues Projects Releases Packages Wiki Activity

fix: store Metabase mappings in ktx.yaml

Browse source
This commit is contained in:
Andrey Avtomonov 2026-05-13 13:08:59 +02:00
parent b9e0a746af
commit 477002805d
23 changed files with 638 additions and 1147 deletions
Download patch file Download diff file Expand all files Collapse all files

33
packages/context/src/ingest/adapters/metabase/fetch.test.ts
View file

@ -36,6 +36,12 @@ function makeMockClient() {
),
getCollectionTree: vi.fn().mockResolvedValue([{ id: 5, name: 'Orders Team', parent_id: null, children: [] }]),
getCollectionItems: vi.fn().mockResolvedValue([]),
getDatabase: vi.fn().mockResolvedValue({
id: 42,
name: 'Analytics',
engine: 'postgres',
details: { host: 'db.example.test', dbname: 'analytics' },
}),
cleanup: vi.fn().mockResolvedValue(undefined),
};
}
@ -253,7 +259,7 @@ describe('fetchMetabaseBundle', () => {
).rejects.toThrow(/mapping.*does not point to connection/);
});
it('throws when the matching mapping has a null metabaseDatabaseName (unhydrated)', async () => {
it('hydrates missing mapping metadata from Metabase instead of requiring a prior refresh', async () => {
sourceStateReader.getSourceState.mockResolvedValue({
syncMode: 'ALL',
selections: [],
@ -268,15 +274,22 @@ describe('fetchMetabaseBundle', () => {
],
defaultTagNames: [],
});
await expect(
fetchMetabaseBundle({
pullConfig: { metabaseConnectionId, metabaseDatabaseId: 42 },
stagedDir,
ctx: makeFetchContext(),
clientFactory,
sourceStateReader,
}),
).rejects.toThrow(/unhydrated.*ktx connection mapping refresh/);
await fetchMetabaseBundle({
pullConfig: { metabaseConnectionId, metabaseDatabaseId: 42 },
stagedDir,
ctx: makeFetchContext(),
clientFactory,
sourceStateReader,
});
expect(clientFactory.__client.getDatabase).toHaveBeenCalledWith(42);
const databaseFile = JSON.parse(await readFile(join(stagedDir, 'databases/42.json'), 'utf-8'));
expect(databaseFile).toMatchObject({
metabaseDatabaseId: 42,
metabaseDatabaseName: 'Analytics',
metabaseEngine: 'postgres',
targetConnectionId,
});
});
it('skips cards whose getResolvedSql returns null and records them in unresolved-cards.json', async () => {

17
packages/context/src/ingest/adapters/metabase/fetch.ts
View file

@ -97,15 +97,16 @@ export async function fetchMetabaseBundle(params: FetchMetabaseBundleParams): Pr
`mapping for database ${pullConfig.metabaseDatabaseId} does not point to connection ${params.ctx.connectionId} (points to ${mapping.targetConnectionId})`,
);
}
if (mapping.metabaseDatabaseName === null) {
throw new IngestInputError(
`mapping for database ${pullConfig.metabaseDatabaseId} on Metabase connection ${pullConfig.metabaseConnectionId} is unhydrated; run \`ktx connection mapping refresh ${pullConfig.metabaseConnectionId}\` to populate metabaseDatabaseName before ingest.`,
);
}
const mappingDatabaseName: string = mapping.metabaseDatabaseName;
const client = await params.clientFactory.createClient(pullConfig, params.ctx);
try {
let mappingDatabaseName = mapping.metabaseDatabaseName;
let mappingEngine = mapping.metabaseEngine;
if (mappingDatabaseName === null) {
const database = await client.getDatabase(pullConfig.metabaseDatabaseId);
mappingDatabaseName = database.name;
mappingEngine = database.engine ?? null;
}
const stagedForScope: StagedSyncConfig = {
metabaseConnectionId: pullConfig.metabaseConnectionId,
metabaseDatabaseId: pullConfig.metabaseDatabaseId,
@ -118,7 +119,7 @@ export async function fetchMetabaseBundle(params: FetchMetabaseBundleParams): Pr
mapping: {
metabaseDatabaseId: mapping.metabaseDatabaseId,
metabaseDatabaseName: mappingDatabaseName,
metabaseEngine: mapping.metabaseEngine,
metabaseEngine: mappingEngine,
targetConnectionId: mapping.targetConnectionId,
},
};
@ -233,7 +234,7 @@ export async function fetchMetabaseBundle(params: FetchMetabaseBundleParams): Pr
const databaseFile: StagedDatabaseFile = {
metabaseDatabaseId: mapping.metabaseDatabaseId,
metabaseDatabaseName: mappingDatabaseName,
metabaseEngine: mapping.metabaseEngine,
metabaseEngine: mappingEngine,
targetConnectionId: mapping.targetConnectionId,
};
await writeFile(

5
packages/context/src/ingest/adapters/metabase/local-metabase.adapter.ts
View file

@ -12,7 +12,7 @@ import {
type MetabaseClientRuntimeConfig,
} from './client-port.js';
import type { MetabaseFetchLogger } from './fetch.js';
import { LocalMetabaseSourceStateReader } from './local-source-state-store.js';
import { KtxYamlMetabaseSourceStateReader, LocalMetabaseDiscoveryCache } from './local-source-state-store.js';
import { MetabaseSourceAdapter } from './metabase.adapter.js';
function stringField(value: unknown): string | null {
@ -62,7 +62,8 @@ export function createLocalMetabaseSourceAdapter(
project: KtxLocalProject,
options: CreateLocalMetabaseSourceAdapterOptions = {},
): MetabaseSourceAdapter {
const sourceStateReader = new LocalMetabaseSourceStateReader({ dbPath: ktxLocalStateDbPath(project) });
const discoveryCache = new LocalMetabaseDiscoveryCache({ dbPath: ktxLocalStateDbPath(project) });
const sourceStateReader = new KtxYamlMetabaseSourceStateReader(project, { discoveryCache });
const connectionFactory = new DefaultMetabaseConnectionClientFactory(
(metabaseConnectionId) =>
metabaseRuntimeConfigFromLocalConnection(

329
packages/context/src/ingest/adapters/metabase/local-source-state-store.test.ts
View file

@ -2,313 +2,112 @@ import { mkdtemp, rm } from 'node:fs/promises';
import { tmpdir } from 'node:os';
import { join } from 'node:path';
import { afterEach, beforeEach, describe, expect, it } from 'vitest';
import { LocalMetabaseSourceStateReader } from './local-source-state-store.js';
import { buildDefaultKtxProjectConfig } from '../../../project/index.js';
import { KtxYamlMetabaseSourceStateReader, LocalMetabaseDiscoveryCache } from './local-source-state-store.js';
describe('LocalMetabaseSourceStateReader', () => {
describe('Metabase YAML source state and discovery cache', () => {
let tempDir: string;
let store: LocalMetabaseSourceStateReader;
let discoveryCache: LocalMetabaseDiscoveryCache;
beforeEach(async () => {
tempDir = await mkdtemp(join(tmpdir(), 'ktx-metabase-local-state-'));
store = new LocalMetabaseSourceStateReader({ dbPath: join(tempDir, '.ktx', 'db.sqlite') });
tempDir = await mkdtemp(join(tmpdir(), 'ktx-metabase-cache-'));
discoveryCache = new LocalMetabaseDiscoveryCache({ dbPath: join(tempDir, '.ktx', 'db.sqlite') });
});
afterEach(async () => {
await rm(tempDir, { recursive: true, force: true });
});
it('round-trips hydrated source state through SQLite', async () => {
await store.replaceSourceState({
connectionId: 'prod-metabase',
function projectWithMetabaseMappings(mappings: Record<string, unknown>) {
return {
config: {
...buildDefaultKtxProjectConfig('metabase-cache-test'),
connections: {
'prod-metabase': {
driver: 'metabase',
mappings,
},
},
},
};
}
it('reads Metabase mapping intent from ktx.yaml config', async () => {
const reader = new KtxYamlMetabaseSourceStateReader(
projectWithMetabaseMappings({
databaseMappings: { '2': 'warehouse' },
syncEnabled: { '2': true },
syncMode: 'ONLY',
selections: { collections: [12], items: [99] },
defaultTagNames: ['analytics'],
}),
{ discoveryCache },
);
await expect(reader.getSourceState('prod-metabase')).resolves.toEqual({
syncMode: 'ONLY',
defaultTagNames: ['analytics', 'curated'],
defaultTagNames: ['analytics'],
selections: [
{ selectionType: 'collection', metabaseObjectId: 10 },
{ selectionType: 'collection', metabaseObjectId: 12 },
{ selectionType: 'item', metabaseObjectId: 99 },
],
mappings: [
{
metabaseDatabaseId: 1,
metabaseDatabaseName: 'Analytics',
metabaseEngine: 'postgres',
metabaseHost: 'warehouse.internal',
metabaseDbName: 'analytics',
targetConnectionId: 'warehouse',
syncEnabled: true,
source: 'cli',
},
],
});
await expect(store.getSourceState('prod-metabase')).resolves.toEqual({
syncMode: 'ONLY',
defaultTagNames: ['analytics', 'curated'],
selections: [
{ selectionType: 'collection', metabaseObjectId: 10 },
{ selectionType: 'item', metabaseObjectId: 99 },
],
mappings: [
{
metabaseDatabaseId: 1,
metabaseDatabaseName: 'Analytics',
metabaseEngine: 'postgres',
targetConnectionId: 'warehouse',
syncEnabled: true,
},
],
});
});
it('excludes unhydrated mappings from getSourceState and exposes them through the side accessor', async () => {
await store.replaceSourceState({
connectionId: 'prod-metabase',
syncMode: 'ALL',
defaultTagNames: [],
selections: [],
mappings: [
{
metabaseDatabaseId: 1,
metabaseDatabaseId: 2,
metabaseDatabaseName: null,
metabaseEngine: null,
metabaseHost: null,
metabaseDbName: null,
targetConnectionId: 'warehouse',
syncEnabled: true,
source: 'ktx.yaml',
},
{
metabaseDatabaseId: 2,
metabaseDatabaseName: 'Sandbox',
metabaseEngine: 'postgres',
metabaseHost: 'warehouse.internal',
metabaseDbName: 'sandbox',
targetConnectionId: 'warehouse',
syncEnabled: true,
source: 'refresh',
},
],
});
const state = await store.getSourceState('prod-metabase');
expect(state.mappings.map((mapping) => mapping.metabaseDatabaseId)).toEqual([2]);
await expect(store.getUnhydratedSyncEnabledMappingIds('prod-metabase')).resolves.toEqual([1]);
});
it('defaults missing sync config to ALL with no tags or selections', async () => {
await store.replaceSourceState({
it('enriches YAML mapping rows with recreatable discovery metadata', async () => {
await discoveryCache.refreshDiscoveredDatabases({
connectionId: 'prod-metabase',
mappings: [
{
metabaseDatabaseId: 3,
metabaseDatabaseName: 'Warehouse',
metabaseEngine: 'postgres',
metabaseHost: null,
metabaseDbName: null,
targetConnectionId: null,
syncEnabled: false,
source: 'refresh',
},
],
discovered: [{ id: 2, name: 'Analytics', engine: 'postgres', host: 'pg.internal', dbName: 'analytics' }],
});
const reader = new KtxYamlMetabaseSourceStateReader(
projectWithMetabaseMappings({
databaseMappings: { '2': 'warehouse' },
syncEnabled: { '2': true },
}),
{ discoveryCache },
);
await expect(store.getSourceState('prod-metabase')).resolves.toMatchObject({
syncMode: 'ALL',
defaultTagNames: [],
selections: [],
});
});
it('supports command-sized mapping writes and reads', async () => {
await store.upsertDatabaseMapping({
connectionId: 'prod-metabase',
metabaseDatabaseId: 1,
targetConnectionId: 'prod-warehouse',
syncEnabled: true,
source: 'cli',
});
await store.setSyncState({
connectionId: 'prod-metabase',
syncMode: 'ONLY',
defaultTagNames: ['analytics'],
selections: [{ selectionType: 'collection', metabaseObjectId: 12 }],
});
await expect(store.listDatabaseMappings('prod-metabase')).resolves.toEqual([
await expect(reader.listDatabaseMappings('prod-metabase')).resolves.toMatchObject([
{
metabaseDatabaseId: 1,
metabaseDatabaseName: null,
metabaseEngine: null,
metabaseHost: null,
metabaseDbName: null,
targetConnectionId: 'prod-warehouse',
syncEnabled: true,
source: 'cli',
},
]);
await expect(store.getUnhydratedSyncEnabledMappingIds('prod-metabase')).resolves.toEqual([1]);
await expect(store.getSourceState('prod-metabase')).resolves.toMatchObject({
syncMode: 'ONLY',
defaultTagNames: ['analytics'],
selections: [{ selectionType: 'collection', metabaseObjectId: 12 }],
mappings: [],
});
});
it('refreshes discovered database metadata while preserving user mapping intent', async () => {
await store.upsertDatabaseMapping({
connectionId: 'prod-metabase',
metabaseDatabaseId: 1,
targetConnectionId: 'prod-warehouse',
syncEnabled: true,
source: 'cli',
});
await store.refreshDiscoveredDatabases({
connectionId: 'prod-metabase',
discovered: [
{ id: 1, name: 'Analytics', engine: 'postgres', host: 'pg.internal', dbName: 'analytics' },
{ id: 2, name: 'Sandbox', engine: 'postgres', host: 'pg.internal', dbName: 'sandbox' },
],
});
await expect(store.listDatabaseMappings('prod-metabase')).resolves.toEqual([
{
metabaseDatabaseId: 1,
metabaseDatabaseId: 2,
metabaseDatabaseName: 'Analytics',
metabaseEngine: 'postgres',
metabaseHost: 'pg.internal',
metabaseDbName: 'analytics',
targetConnectionId: 'prod-warehouse',
targetConnectionId: 'warehouse',
syncEnabled: true,
source: 'cli',
source: 'ktx.yaml',
},
]);
});
it('lists discovered-only rows as refresh cache data without turning them into config state', async () => {
await discoveryCache.refreshDiscoveredDatabases({
connectionId: 'prod-metabase',
discovered: [{ id: 7, name: 'Unmapped', engine: 'mysql', host: 'mysql.internal', dbName: 'sales' }],
});
const reader = new KtxYamlMetabaseSourceStateReader(projectWithMetabaseMappings({}), { discoveryCache });
await expect(reader.getSourceState('prod-metabase')).resolves.toMatchObject({ mappings: [] });
await expect(reader.listDatabaseMappings('prod-metabase')).resolves.toMatchObject([
{
metabaseDatabaseId: 2,
metabaseDatabaseName: 'Sandbox',
metabaseEngine: 'postgres',
metabaseHost: 'pg.internal',
metabaseDbName: 'sandbox',
metabaseDatabaseId: 7,
metabaseDatabaseName: 'Unmapped',
targetConnectionId: null,
syncEnabled: false,
source: 'refresh',
},
]);
});
it('updates sync-enabled, clears scoped rows, and applies bulk state in one call', async () => {
await store.replaceSourceState({
connectionId: 'prod-metabase',
mappings: [
{
metabaseDatabaseId: 1,
metabaseDatabaseName: 'Analytics',
metabaseEngine: 'postgres',
metabaseHost: 'pg.internal',
metabaseDbName: 'analytics',
targetConnectionId: 'prod-warehouse',
syncEnabled: true,
source: 'refresh',
},
{
metabaseDatabaseId: 2,
metabaseDatabaseName: 'Sandbox',
metabaseEngine: 'postgres',
metabaseHost: 'pg.internal',
metabaseDbName: 'sandbox',
targetConnectionId: 'staging-warehouse',
syncEnabled: true,
source: 'refresh',
},
],
});
await store.setMappingSyncEnabled({
connectionId: 'prod-metabase',
metabaseDatabaseId: 2,
syncEnabled: false,
});
await store.clearDatabaseMappings({ connectionId: 'prod-metabase', metabaseDatabaseId: 1 });
await expect(store.listDatabaseMappings('prod-metabase')).resolves.toEqual([
{
metabaseDatabaseId: 2,
metabaseDatabaseName: 'Sandbox',
metabaseEngine: 'postgres',
metabaseHost: 'pg.internal',
metabaseDbName: 'sandbox',
targetConnectionId: 'staging-warehouse',
syncEnabled: false,
source: 'refresh',
},
]);
});
it('seeds unhydrated yaml intent without exposing it through getSourceState', async () => {
await store.applyYamlBootstrap({
connectionId: 'prod-metabase',
syncMode: 'ALL',
defaultTagNames: ['ktx'],
selections: [{ selectionType: 'collection', metabaseObjectId: 12 }],
mappings: [{ metabaseDatabaseId: 1, targetConnectionId: 'prod-warehouse', syncEnabled: true }],
});
await expect(store.getUnhydratedSyncEnabledMappingIds('prod-metabase')).resolves.toEqual([1]);
await expect(store.getSourceState('prod-metabase')).resolves.toMatchObject({
syncMode: 'ALL',
defaultTagNames: ['ktx'],
selections: [{ selectionType: 'collection', metabaseObjectId: 12 }],
mappings: [],
});
await expect(store.listDatabaseMappings('prod-metabase')).resolves.toMatchObject([
{
metabaseDatabaseId: 1,
metabaseDatabaseName: null,
targetConnectionId: 'prod-warehouse',
syncEnabled: true,
source: 'ktx.yaml',
},
]);
});
it('applies yaml target intent onto refresh metadata but does not overwrite cli rows', async () => {
await store.refreshDiscoveredDatabases({
connectionId: 'prod-metabase',
discovered: [{ id: 1, name: 'Analytics', engine: 'postgres', host: 'db.test', dbName: 'analytics' }],
});
await store.upsertDatabaseMapping({
connectionId: 'prod-metabase',
metabaseDatabaseId: 2,
targetConnectionId: 'cli-warehouse',
syncEnabled: true,
source: 'cli',
});
await store.applyYamlBootstrap({
connectionId: 'prod-metabase',
syncMode: 'EXCEPT',
defaultTagNames: [],
selections: [{ selectionType: 'item', metabaseObjectId: 99 }],
mappings: [
{ metabaseDatabaseId: 1, targetConnectionId: 'yaml-warehouse', syncEnabled: true },
{ metabaseDatabaseId: 2, targetConnectionId: 'yaml-warehouse', syncEnabled: false },
],
});
await expect(store.listDatabaseMappings('prod-metabase')).resolves.toMatchObject([
{
metabaseDatabaseId: 1,
metabaseDatabaseName: 'Analytics',
metabaseEngine: 'postgres',
targetConnectionId: 'yaml-warehouse',
syncEnabled: true,
source: 'ktx.yaml',
},
{
metabaseDatabaseId: 2,
targetConnectionId: 'cli-warehouse',
syncEnabled: true,
source: 'cli',
},
]);
});
});

623
packages/context/src/ingest/adapters/metabase/local-source-state-store.ts
View file

@ -1,17 +1,31 @@
import { mkdirSync } from 'node:fs';
import { dirname } from 'node:path';
import Database from 'better-sqlite3';
import {
parseMetabaseMappingBootstrap,
type KtxLocalProject,
type MetabaseMappingBootstrap,
} from '../../../project/index.js';
import type { DiscoveredMetabaseDatabase } from './mapping.js';
import type { MetabaseSourceState, MetabaseSourceStateReader, MetabaseSourceStateSelection } from './source-state-port.js';
import type { MetabaseSyncMode } from './types.js';
export type LocalMetabaseMappingSource = 'ktx.yaml' | 'cli' | 'refresh';
export type LocalMetabaseMappingSource = 'ktx.yaml' | 'refresh';
interface LocalMetabaseSourceStateStoreOptions {
interface LocalMetabaseDiscoveryCacheOptions {
dbPath: string;
now?: () => Date;
}
export interface LocalMetabaseSourceStateMappingInput {
export interface RefreshLocalMetabaseDiscoveredDatabasesInput {
connectionId: string;
discovered: DiscoveredMetabaseDatabase[];
}
export interface LocalMetabaseDiscoveredDatabaseRow extends DiscoveredMetabaseDatabase {
updatedAt: string;
}
export interface LocalMetabaseMappingListRow {
metabaseDatabaseId: number;
metabaseDatabaseName: string | null;
metabaseEngine: string | null;
@ -22,443 +36,86 @@ export interface LocalMetabaseSourceStateMappingInput {
source: LocalMetabaseMappingSource;
}
export interface ReplaceLocalMetabaseSourceStateInput {
connectionId: string;
syncMode?: MetabaseSyncMode;
defaultTagNames?: string[];
selections?: MetabaseSourceStateSelection[];
mappings: LocalMetabaseSourceStateMappingInput[];
}
interface ApplyLocalMetabaseYamlBootstrapInput {
connectionId: string;
syncMode: MetabaseSyncMode;
defaultTagNames: string[];
selections: MetabaseSourceStateSelection[];
mappings: Array<{
metabaseDatabaseId: number;
targetConnectionId: string | null;
syncEnabled: boolean;
}>;
}
export interface LocalMetabaseMappingListRow extends LocalMetabaseSourceStateMappingInput {}
export interface UpsertLocalMetabaseDatabaseMappingInput {
connectionId: string;
metabaseDatabaseId: number;
targetConnectionId: string | null;
syncEnabled: boolean;
source: LocalMetabaseMappingSource;
}
export interface SetLocalMetabaseMappingSyncEnabledInput {
connectionId: string;
metabaseDatabaseId: number;
syncEnabled: boolean;
}
export interface SetLocalMetabaseSyncStateInput {
connectionId: string;
syncMode: MetabaseSyncMode;
defaultTagNames: string[];
selections: MetabaseSourceStateSelection[];
}
export interface RefreshLocalMetabaseDiscoveredDatabasesInput {
connectionId: string;
discovered: Array<{
id: number;
name: string;
engine: string;
host: string | null;
dbName: string | null;
}>;
}
export interface ClearLocalMetabaseMappingsInput {
connectionId: string;
metabaseDatabaseId?: number;
}
interface SelectionRow {
selection_type: 'collection' | 'item';
metabase_object_id: number;
}
interface MappingRow {
interface DiscoveryRow {
metabase_database_id: number;
metabase_database_name: string | null;
metabase_engine: string | null;
target_connection_id: string | null;
sync_enabled: number;
metabase_database_name: string;
metabase_engine: string;
metabase_host: string | null;
metabase_db_name: string | null;
updated_at: string;
}
interface SyncConfigRow {
sync_mode: MetabaseSyncMode;
default_tag_names_json: string;
function selectionState(bootstrap: MetabaseMappingBootstrap): MetabaseSourceStateSelection[] {
return [
...bootstrap.selections.collections.map((id) => ({ selectionType: 'collection' as const, metabaseObjectId: id })),
...bootstrap.selections.items.map((id) => ({ selectionType: 'item' as const, metabaseObjectId: id })),
];
}
function parseDefaultTagNames(raw: string): string[] {
const parsed = JSON.parse(raw);
return Array.isArray(parsed) ? parsed.filter((value): value is string => typeof value === 'string') : [];
function configuredMappingIds(bootstrap: MetabaseMappingBootstrap): number[] {
return [...new Set([...Object.keys(bootstrap.databaseMappings), ...Object.keys(bootstrap.syncEnabled)].map(Number))].sort(
(left, right) => left - right,
);
}
export class LocalMetabaseSourceStateReader implements MetabaseSourceStateReader {
function discoveredRowToDatabase(row: DiscoveryRow): LocalMetabaseDiscoveredDatabaseRow {
return {
id: row.metabase_database_id,
name: row.metabase_database_name,
engine: row.metabase_engine,
host: row.metabase_host,
dbName: row.metabase_db_name,
updatedAt: row.updated_at,
};
}
function emptyMetabaseSourceState(): MetabaseSourceState {
return {
syncMode: 'ALL',
selections: [],
defaultTagNames: [],
mappings: [],
};
}
export class LocalMetabaseDiscoveryCache {
private readonly db: Database.Database;
private readonly now: () => Date;
constructor(options: LocalMetabaseSourceStateStoreOptions) {
constructor(options: LocalMetabaseDiscoveryCacheOptions) {
mkdirSync(dirname(options.dbPath), { recursive: true });
this.db = new Database(options.dbPath);
this.db.pragma('journal_mode = WAL');
this.db.pragma('foreign_keys = ON');
this.now = options.now ?? (() => new Date());
this.db.exec(`
CREATE TABLE IF NOT EXISTS local_metabase_sync_config (
metabase_connection_id TEXT PRIMARY KEY,
sync_mode TEXT NOT NULL,
default_tag_names_json TEXT NOT NULL,
updated_at TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS local_metabase_selections (
metabase_connection_id TEXT NOT NULL,
selection_type TEXT NOT NULL,
metabase_object_id INTEGER NOT NULL,
PRIMARY KEY (metabase_connection_id, selection_type, metabase_object_id)
);
CREATE TABLE IF NOT EXISTS local_metabase_database_mappings (
CREATE TABLE IF NOT EXISTS local_metabase_discovered_databases (
metabase_connection_id TEXT NOT NULL,
metabase_database_id INTEGER NOT NULL,
metabase_database_name TEXT,
metabase_engine TEXT,
metabase_database_name TEXT NOT NULL,
metabase_engine TEXT NOT NULL,
metabase_host TEXT,
metabase_db_name TEXT,
target_connection_id TEXT,
sync_enabled INTEGER NOT NULL DEFAULT 0,
source TEXT NOT NULL,
updated_at TEXT NOT NULL,
PRIMARY KEY (metabase_connection_id, metabase_database_id)
);
`);
}
async applyYamlBootstrap(input: ApplyLocalMetabaseYamlBootstrapInput): Promise<void> {
const timestamp = this.now().toISOString();
const apply = this.db.transaction(() => {
const syncConfigExists = this.db
.prepare('SELECT 1 FROM local_metabase_sync_config WHERE metabase_connection_id = ?')
.get(input.connectionId);
if (!syncConfigExists) {
this.db
.prepare(
`
INSERT INTO local_metabase_sync_config (
metabase_connection_id,
sync_mode,
default_tag_names_json,
updated_at
)
VALUES (?, ?, ?, ?)
`,
)
.run(input.connectionId, input.syncMode, JSON.stringify(input.defaultTagNames), timestamp);
const insertSelection = this.db.prepare(`
INSERT INTO local_metabase_selections (
metabase_connection_id,
selection_type,
metabase_object_id
)
VALUES (?, ?, ?)
`);
for (const selection of input.selections) {
insertSelection.run(input.connectionId, selection.selectionType, selection.metabaseObjectId);
}
}
const existing = this.db.prepare(`
SELECT target_connection_id, source
FROM local_metabase_database_mappings
WHERE metabase_connection_id = ? AND metabase_database_id = ?
`);
const insert = this.db.prepare(`
INSERT INTO local_metabase_database_mappings (
metabase_connection_id,
metabase_database_id,
metabase_database_name,
metabase_engine,
metabase_host,
metabase_db_name,
target_connection_id,
sync_enabled,
source,
updated_at
)
VALUES (?, ?, NULL, NULL, NULL, NULL, ?, ?, 'ktx.yaml', ?)
`);
const updateRefreshRow = this.db.prepare(`
UPDATE local_metabase_database_mappings
SET target_connection_id = ?,
sync_enabled = ?,
source = 'ktx.yaml',
updated_at = ?
WHERE metabase_connection_id = ?
AND metabase_database_id = ?
AND source = 'refresh'
AND target_connection_id IS NULL
`);
for (const mapping of input.mappings) {
const row = existing.get(input.connectionId, mapping.metabaseDatabaseId) as
| { target_connection_id: string | null; source: LocalMetabaseMappingSource }
| undefined;
if (!row) {
insert.run(
input.connectionId,
mapping.metabaseDatabaseId,
mapping.targetConnectionId,
mapping.syncEnabled ? 1 : 0,
timestamp,
);
continue;
}
if (row.source === 'refresh' && row.target_connection_id === null) {
updateRefreshRow.run(
mapping.targetConnectionId,
mapping.syncEnabled ? 1 : 0,
timestamp,
input.connectionId,
mapping.metabaseDatabaseId,
);
}
}
});
apply();
}
async replaceSourceState(input: ReplaceLocalMetabaseSourceStateInput): Promise<void> {
const timestamp = this.now().toISOString();
const syncMode = input.syncMode ?? 'ALL';
const selections = input.selections ?? [];
const defaultTagNames = input.defaultTagNames ?? [];
const replace = this.db.transaction(() => {
this.db
.prepare(
`
INSERT INTO local_metabase_sync_config (
metabase_connection_id,
sync_mode,
default_tag_names_json,
updated_at
)
VALUES (?, ?, ?, ?)
ON CONFLICT(metabase_connection_id) DO UPDATE SET
sync_mode = excluded.sync_mode,
default_tag_names_json = excluded.default_tag_names_json,
updated_at = excluded.updated_at
`,
)
.run(input.connectionId, syncMode, JSON.stringify(defaultTagNames), timestamp);
this.db.prepare('DELETE FROM local_metabase_selections WHERE metabase_connection_id = ?').run(input.connectionId);
const insertSelection = this.db.prepare(`
INSERT INTO local_metabase_selections (
metabase_connection_id,
selection_type,
metabase_object_id
)
VALUES (?, ?, ?)
`);
for (const selection of selections) {
insertSelection.run(input.connectionId, selection.selectionType, selection.metabaseObjectId);
}
this.db
.prepare('DELETE FROM local_metabase_database_mappings WHERE metabase_connection_id = ?')
.run(input.connectionId);
const insertMapping = this.db.prepare(`
INSERT INTO local_metabase_database_mappings (
metabase_connection_id,
metabase_database_id,
metabase_database_name,
metabase_engine,
metabase_host,
metabase_db_name,
target_connection_id,
sync_enabled,
source,
updated_at
)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
`);
for (const mapping of input.mappings) {
insertMapping.run(
input.connectionId,
mapping.metabaseDatabaseId,
mapping.metabaseDatabaseName,
mapping.metabaseEngine,
mapping.metabaseHost,
mapping.metabaseDbName,
mapping.targetConnectionId,
mapping.syncEnabled ? 1 : 0,
mapping.source,
timestamp,
);
}
});
replace();
}
async listDatabaseMappings(connectionId: string): Promise<LocalMetabaseMappingListRow[]> {
const rows = this.db
.prepare(
`
SELECT
metabase_database_id,
metabase_database_name,
metabase_engine,
metabase_host,
metabase_db_name,
target_connection_id,
sync_enabled,
source
FROM local_metabase_database_mappings
WHERE metabase_connection_id = ?
ORDER BY metabase_database_id
`,
)
.all(connectionId) as Array<{
metabase_database_id: number;
metabase_database_name: string | null;
metabase_engine: string | null;
metabase_host: string | null;
metabase_db_name: string | null;
target_connection_id: string | null;
sync_enabled: number;
source: LocalMetabaseMappingSource;
}>;
return rows.map((row) => ({
metabaseDatabaseId: row.metabase_database_id,
metabaseDatabaseName: row.metabase_database_name,
metabaseEngine: row.metabase_engine,
metabaseHost: row.metabase_host,
metabaseDbName: row.metabase_db_name,
targetConnectionId: row.target_connection_id,
syncEnabled: row.sync_enabled === 1,
source: row.source,
}));
}
async upsertDatabaseMapping(input: UpsertLocalMetabaseDatabaseMappingInput): Promise<void> {
const timestamp = this.now().toISOString();
this.db
.prepare(
`
INSERT INTO local_metabase_database_mappings (
metabase_connection_id,
metabase_database_id,
metabase_database_name,
metabase_engine,
metabase_host,
metabase_db_name,
target_connection_id,
sync_enabled,
source,
updated_at
)
VALUES (?, ?, NULL, NULL, NULL, NULL, ?, ?, ?, ?)
ON CONFLICT(metabase_connection_id, metabase_database_id) DO UPDATE SET
target_connection_id = excluded.target_connection_id,
sync_enabled = excluded.sync_enabled,
source = excluded.source,
updated_at = excluded.updated_at
`,
)
.run(
input.connectionId,
input.metabaseDatabaseId,
input.targetConnectionId,
input.syncEnabled ? 1 : 0,
input.source,
timestamp,
);
}
async setMappingSyncEnabled(input: SetLocalMetabaseMappingSyncEnabledInput): Promise<void> {
const timestamp = this.now().toISOString();
this.db
.prepare(
`
UPDATE local_metabase_database_mappings
SET sync_enabled = ?, updated_at = ?
WHERE metabase_connection_id = ? AND metabase_database_id = ?
`,
)
.run(input.syncEnabled ? 1 : 0, timestamp, input.connectionId, input.metabaseDatabaseId);
}
async setSyncState(input: SetLocalMetabaseSyncStateInput): Promise<void> {
const timestamp = this.now().toISOString();
const write = this.db.transaction(() => {
this.db
.prepare(
`
INSERT INTO local_metabase_sync_config (
metabase_connection_id,
sync_mode,
default_tag_names_json,
updated_at
)
VALUES (?, ?, ?, ?)
ON CONFLICT(metabase_connection_id) DO UPDATE SET
sync_mode = excluded.sync_mode,
default_tag_names_json = excluded.default_tag_names_json,
updated_at = excluded.updated_at
`,
)
.run(input.connectionId, input.syncMode, JSON.stringify(input.defaultTagNames), timestamp);
this.db.prepare('DELETE FROM local_metabase_selections WHERE metabase_connection_id = ?').run(input.connectionId);
const insertSelection = this.db.prepare(`
INSERT INTO local_metabase_selections (
metabase_connection_id,
selection_type,
metabase_object_id
)
VALUES (?, ?, ?)
`);
for (const selection of input.selections) {
insertSelection.run(input.connectionId, selection.selectionType, selection.metabaseObjectId);
}
});
write();
}
async refreshDiscoveredDatabases(input: RefreshLocalMetabaseDiscoveredDatabasesInput): Promise<void> {
const timestamp = this.now().toISOString();
const refresh = this.db.transaction(() => {
const upsert = this.db.prepare(`
INSERT INTO local_metabase_database_mappings (
INSERT INTO local_metabase_discovered_databases (
metabase_connection_id,
metabase_database_id,
metabase_database_name,
metabase_engine,
metabase_host,
metabase_db_name,
target_connection_id,
sync_enabled,
source,
updated_at
)
VALUES (?, ?, ?, ?, ?, ?, NULL, 0, 'refresh', ?)
VALUES (?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(metabase_connection_id, metabase_database_id) DO UPDATE SET
metabase_database_name = excluded.metabase_database_name,
metabase_engine = excluded.metabase_engine,
@ -483,78 +140,116 @@ export class LocalMetabaseSourceStateReader implements MetabaseSourceStateReader
refresh();
}
async clearDatabaseMappings(input: ClearLocalMetabaseMappingsInput): Promise<void> {
if (input.metabaseDatabaseId === undefined) {
this.db.prepare('DELETE FROM local_metabase_database_mappings WHERE metabase_connection_id = ?').run(input.connectionId);
return;
}
this.db
.prepare('DELETE FROM local_metabase_database_mappings WHERE metabase_connection_id = ? AND metabase_database_id = ?')
.run(input.connectionId, input.metabaseDatabaseId);
}
async getUnhydratedSyncEnabledMappingIds(connectionId: string): Promise<number[]> {
async listDiscoveredDatabases(connectionId: string): Promise<LocalMetabaseDiscoveredDatabaseRow[]> {
const rows = this.db
.prepare(
`
SELECT metabase_database_id
FROM local_metabase_database_mappings
WHERE metabase_connection_id = ?
AND sync_enabled = 1
AND target_connection_id IS NOT NULL
AND metabase_database_name IS NULL
ORDER BY metabase_database_id
`,
)
.all(connectionId) as Array<{ metabase_database_id: number }>;
return rows.map((row) => row.metabase_database_id);
}
async getSourceState(connectionId: string): Promise<MetabaseSourceState> {
const config = this.db
.prepare('SELECT sync_mode, default_tag_names_json FROM local_metabase_sync_config WHERE metabase_connection_id = ?')
.get(connectionId) as SyncConfigRow | undefined;
const selections = this.db
.prepare(
`
SELECT selection_type, metabase_object_id
FROM local_metabase_selections
WHERE metabase_connection_id = ?
ORDER BY selection_type, metabase_object_id
`,
)
.all(connectionId) as SelectionRow[];
const mappings = this.db
.prepare(
`
SELECT
metabase_database_id,
metabase_database_name,
metabase_engine,
target_connection_id,
sync_enabled
FROM local_metabase_database_mappings
metabase_host,
metabase_db_name,
updated_at
FROM local_metabase_discovered_databases
WHERE metabase_connection_id = ?
AND metabase_database_name IS NOT NULL
ORDER BY metabase_database_id
`,
)
.all(connectionId) as MappingRow[];
.all(connectionId) as DiscoveryRow[];
return rows.map(discoveredRowToDatabase);
}
return {
syncMode: config?.sync_mode ?? 'ALL',
defaultTagNames: config ? parseDefaultTagNames(config.default_tag_names_json) : [],
selections: selections.map((selection) => ({
selectionType: selection.selection_type,
metabaseObjectId: selection.metabase_object_id,
})),
mappings: mappings.map((mapping) => ({
metabaseDatabaseId: mapping.metabase_database_id,
metabaseDatabaseName: mapping.metabase_database_name,
metabaseEngine: mapping.metabase_engine,
targetConnectionId: mapping.target_connection_id,
syncEnabled: mapping.sync_enabled === 1,
})),
};
async getDiscoveredDatabase(
connectionId: string,
metabaseDatabaseId: number,
): Promise<LocalMetabaseDiscoveredDatabaseRow | null> {
const row = this.db
.prepare(
`
SELECT
metabase_database_id,
metabase_database_name,
metabase_engine,
metabase_host,
metabase_db_name,
updated_at
FROM local_metabase_discovered_databases
WHERE metabase_connection_id = ? AND metabase_database_id = ?
`,
)
.get(connectionId, metabaseDatabaseId) as DiscoveryRow | undefined;
return row ? discoveredRowToDatabase(row) : null;
}
}
export class KtxYamlMetabaseSourceStateReader implements MetabaseSourceStateReader {
constructor(
private readonly project: Pick<KtxLocalProject, 'config'>,
private readonly options: { discoveryCache?: LocalMetabaseDiscoveryCache } = {},
) {}
async getSourceState(connectionId: string): Promise<MetabaseSourceState> {
const connection = this.project.config.connections[connectionId];
if (!connection || String(connection.driver ?? '').toLowerCase() !== 'metabase') {
return emptyMetabaseSourceState();
}
const bootstrap = parseMetabaseMappingBootstrap(connectionId, connection);
const discovered = new Map(
(await this.options.discoveryCache?.listDiscoveredDatabases(connectionId))?.map((database) => [database.id, database]) ??
[],
);
return {
syncMode: bootstrap.syncMode,
selections: selectionState(bootstrap),
defaultTagNames: bootstrap.defaultTagNames,
mappings: configuredMappingIds(bootstrap).map((id) => {
const metadata = discovered.get(id);
return {
metabaseDatabaseId: id,
metabaseDatabaseName: metadata?.name ?? null,
metabaseEngine: metadata?.engine ?? null,
metabaseHost: metadata?.host ?? null,
metabaseDbName: metadata?.dbName ?? null,
targetConnectionId: bootstrap.databaseMappings[String(id)] ?? null,
syncEnabled: bootstrap.syncEnabled[String(id)] ?? false,
};
}),
};
}
async listDatabaseMappings(connectionId: string): Promise<LocalMetabaseMappingListRow[]> {
const state = await this.getSourceState(connectionId);
const configuredRows: LocalMetabaseMappingListRow[] = state.mappings.map((mapping) => ({
metabaseDatabaseId: mapping.metabaseDatabaseId,
metabaseDatabaseName: mapping.metabaseDatabaseName,
metabaseEngine: mapping.metabaseEngine,
metabaseHost: mapping.metabaseHost ?? null,
metabaseDbName: mapping.metabaseDbName ?? null,
targetConnectionId: mapping.targetConnectionId,
syncEnabled: mapping.syncEnabled,
source: 'ktx.yaml',
}));
const configuredIds = new Set(configuredRows.map((row) => row.metabaseDatabaseId));
const discoveredRows =
(await this.options.discoveryCache?.listDiscoveredDatabases(connectionId))?.filter(
(database) => !configuredIds.has(database.id),
) ?? [];
return [
...configuredRows,
...discoveredRows.map((database) => ({
metabaseDatabaseId: database.id,
metabaseDatabaseName: database.name,
metabaseEngine: database.engine,
metabaseHost: database.host,
metabaseDbName: database.dbName,
targetConnectionId: null,
syncEnabled: false,
source: 'refresh' as const,
})),
].sort((left, right) => left.metabaseDatabaseId - right.metabaseDatabaseId);
}
}

2
packages/context/src/ingest/adapters/metabase/source-state-port.ts
View file

@ -9,6 +9,8 @@ export interface MetabaseSourceStateMapping {
metabaseDatabaseId: number;
metabaseDatabaseName: string | null;
metabaseEngine: string | null;
metabaseHost?: string | null;
metabaseDbName?: string | null;
targetConnectionId: string | null;
syncEnabled: boolean;
}

12
packages/context/src/ingest/index.ts
View file

@ -240,17 +240,15 @@ export {
createLocalMetabaseSourceAdapter,
metabaseRuntimeConfigFromLocalConnection,
} from './adapters/metabase/local-metabase.adapter.js';
export { LocalMetabaseSourceStateReader } from './adapters/metabase/local-source-state-store.js';
export {
KtxYamlMetabaseSourceStateReader,
LocalMetabaseDiscoveryCache,
} from './adapters/metabase/local-source-state-store.js';
export type {
ClearLocalMetabaseMappingsInput,
LocalMetabaseDiscoveredDatabaseRow,
LocalMetabaseMappingListRow,
LocalMetabaseMappingSource,
LocalMetabaseSourceStateMappingInput,
ReplaceLocalMetabaseSourceStateInput,
RefreshLocalMetabaseDiscoveredDatabasesInput,
SetLocalMetabaseMappingSyncEnabledInput,
SetLocalMetabaseSyncStateInput,
UpsertLocalMetabaseDatabaseMappingInput,
} from './adapters/metabase/local-source-state-store.js';
export { metabaseLocalConnectionIdSchema, metabasePullConfigSchema, parseMetabasePullConfig } from './adapters/metabase/types.js';
export type { MetabasePullConfig, MetabaseSyncMode } from './adapters/metabase/types.js';

15
packages/context/src/ingest/local-ingest.ts
View file

@ -9,10 +9,9 @@ import type { KtxLocalProject } from '../project/index.js';
import { ktxLocalStateDbPath } from '../project/index.js';
import type { KtxQueryResult } from '../sl/index.js';
import { planMetabaseFanoutChildren } from './adapters/metabase/fanout-planner.js';
import { LocalMetabaseSourceStateReader } from './adapters/metabase/local-source-state-store.js';
import { KtxYamlMetabaseSourceStateReader, LocalMetabaseDiscoveryCache } from './adapters/metabase/local-source-state-store.js';
import { localPullConfigForAdapter, type DefaultLocalIngestAdaptersOptions } from './local-adapters.js';
import { createLocalBundleIngestRuntime } from './local-bundle-runtime.js';
import { seedLocalMappingStateFromKtxYaml } from './local-mapping-reconcile.js';
import type { MemoryFlowEventSink } from './memory-flow/types.js';
import { buildSyncId } from './raw-sources-paths.js';
import type { IngestReportBody, IngestReportSnapshot } from './reports.js';
@ -364,16 +363,10 @@ export async function runLocalMetabaseIngest(
const metabaseConnectionId = safeSegment('metabase connection id', options.metabaseConnectionId);
assertConfigured(options.project, 'metabase', metabaseConnectionId);
await seedLocalMappingStateFromKtxYaml(options.project, metabaseConnectionId);
const adapter = findAdapter(options.adapters, 'metabase');
const sourceStateReader = new LocalMetabaseSourceStateReader({ dbPath: ktxLocalStateDbPath(options.project) });
const unhydrated = await sourceStateReader.getUnhydratedSyncEnabledMappingIds(metabaseConnectionId);
if (unhydrated.length > 0) {
throw new Error(
`Metabase mappings ${unhydrated.join(', ')} are not hydrated; run \`ktx connection mapping refresh ${metabaseConnectionId}\` before local Metabase ingest.`,
);
}
const sourceStateReader = new KtxYamlMetabaseSourceStateReader(options.project, {
discoveryCache: new LocalMetabaseDiscoveryCache({ dbPath: ktxLocalStateDbPath(options.project) }),
});
const state = await sourceStateReader.getSourceState(metabaseConnectionId);
const childPlans = planMetabaseFanoutChildren({

15
packages/context/src/ingest/local-mapping-reconcile.test.ts
View file

@ -4,7 +4,6 @@ import { join } from 'node:path';
import { afterEach, describe, expect, it } from 'vitest';
import { ktxLocalStateDbPath, type KtxLocalProject } from '../project/index.js';
import { LocalLookerRuntimeStore } from './adapters/looker/local-runtime-store.js';
import { LocalMetabaseSourceStateReader } from './adapters/metabase/local-source-state-store.js';
import { seedLocalMappingStateFromKtxYaml } from './local-mapping-reconcile.js';
describe('local mapping yaml reconciliation bridge', () => {
@ -23,7 +22,7 @@ describe('local mapping yaml reconciliation bridge', () => {
} as KtxLocalProject;
}
it('seeds Metabase local state from ktx.yaml mapping intent', async () => {
it('does not copy Metabase mapping intent into local SQLite state', async () => {
tempDir = await mkdtemp(join(tmpdir(), 'ktx-metabase-yaml-seed-'));
const project = projectWithConnections({
'prod-metabase': {
@ -39,17 +38,7 @@ describe('local mapping yaml reconciliation bridge', () => {
'prod-warehouse': { driver: 'postgres', url: 'postgresql://readonly@db.test/analytics' },
});
await seedLocalMappingStateFromKtxYaml(project, 'prod-metabase');
const store = new LocalMetabaseSourceStateReader({ dbPath: ktxLocalStateDbPath(project) });
await expect(store.listDatabaseMappings('prod-metabase')).resolves.toMatchObject([
{ metabaseDatabaseId: 1, targetConnectionId: 'prod-warehouse', syncEnabled: true, source: 'ktx.yaml' },
]);
await expect(store.getSourceState('prod-metabase')).resolves.toMatchObject({
syncMode: 'ONLY',
selections: [{ selectionType: 'collection', metabaseObjectId: 12 }],
defaultTagNames: ['ktx'],
});
await expect(seedLocalMappingStateFromKtxYaml(project, 'prod-metabase')).resolves.toBeUndefined();
});
it('seeds Looker local mappings from ktx.yaml mapping intent', async () => {

31
packages/context/src/ingest/local-mapping-reconcile.ts
View file

@ -3,29 +3,8 @@ import {
parseConnectionMappingBootstrap,
type KtxLocalProject,
type LookerMappingBootstrap,
type MetabaseMappingBootstrap,
} from '../project/index.js';
import { LocalLookerRuntimeStore } from './adapters/looker/local-runtime-store.js';
import { LocalMetabaseSourceStateReader } from './adapters/metabase/local-source-state-store.js';
function metabaseSelections(bootstrap: MetabaseMappingBootstrap) {
return [
...bootstrap.selections.collections.map((id) => ({ selectionType: 'collection' as const, metabaseObjectId: id })),
...bootstrap.selections.items.map((id) => ({ selectionType: 'item' as const, metabaseObjectId: id })),
];
}
function metabaseMappings(bootstrap: MetabaseMappingBootstrap) {
const ids = new Set([...Object.keys(bootstrap.databaseMappings), ...Object.keys(bootstrap.syncEnabled)]);
return [...ids]
.map((id) => Number(id))
.sort((a, b) => a - b)
.map((id) => ({
metabaseDatabaseId: id,
targetConnectionId: bootstrap.databaseMappings[String(id)] ?? null,
syncEnabled: bootstrap.syncEnabled[String(id)] ?? false,
}));
}
function lookerMappings(bootstrap: LookerMappingBootstrap) {
return Object.entries(bootstrap.connectionMappings)
@ -44,20 +23,12 @@ export async function seedLocalMappingStateFromKtxYaml(project: KtxLocalProject,
return;
}
const dbPath = ktxLocalStateDbPath(project);
if (bootstrap.adapter === 'metabase') {
await new LocalMetabaseSourceStateReader({ dbPath }).applyYamlBootstrap({
connectionId,
syncMode: bootstrap.syncMode,
defaultTagNames: bootstrap.defaultTagNames,
selections: metabaseSelections(bootstrap),
mappings: metabaseMappings(bootstrap),
});
return;
}
if (bootstrap.adapter === 'looker') {
await new LocalLookerRuntimeStore({ dbPath }).applyYamlBootstrap({
await new LocalLookerRuntimeStore({ dbPath: ktxLocalStateDbPath(project) }).applyYamlBootstrap({
lookerConnectionId: connectionId,
mappings: lookerMappings(bootstrap),
});

143
packages/context/src/ingest/local-metabase-ingest.test.ts
View file

@ -4,7 +4,7 @@ import { join } from 'node:path';
import { AgentRunnerService } from '../agent/index.js';
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
import { initKtxProject, type KtxLocalProject } from '../project/index.js';
import { LocalMetabaseSourceStateReader } from './adapters/metabase/local-source-state-store.js';
import { LocalMetabaseDiscoveryCache } from './adapters/metabase/local-source-state-store.js';
import { getLocalIngestStatus, runLocalMetabaseIngest } from './local-ingest.js';
import type { ChunkResult, FetchContext, SourceAdapter } from './types.js';
@ -94,33 +94,19 @@ describe('runLocalMetabaseIngest', () => {
});
async function seedMetabaseState(): Promise<void> {
const store = new LocalMetabaseSourceStateReader({ dbPath: join(tempDir, '.ktx', 'db.sqlite') });
await store.replaceSourceState({
connectionId: 'prod-metabase',
project.config.connections['prod-metabase'].mappings = {
databaseMappings: { '1': 'warehouse_a', '2': 'warehouse_b' },
syncEnabled: { '1': true, '2': true },
syncMode: 'ALL',
defaultTagNames: ['ktx'],
selections: [],
mappings: [
{
metabaseDatabaseId: 1,
metabaseDatabaseName: 'Warehouse A',
metabaseEngine: 'postgres',
metabaseHost: 'localhost',
metabaseDbName: 'a',
targetConnectionId: 'warehouse_a',
syncEnabled: true,
source: 'refresh',
},
{
metabaseDatabaseId: 2,
metabaseDatabaseName: 'Warehouse B',
metabaseEngine: 'postgres',
metabaseHost: 'localhost',
metabaseDbName: 'b',
targetConnectionId: 'warehouse_b',
syncEnabled: true,
source: 'refresh',
},
selections: { collections: [], items: [] },
};
const discoveryCache = new LocalMetabaseDiscoveryCache({ dbPath: join(tempDir, '.ktx', 'db.sqlite') });
await discoveryCache.refreshDiscoveredDatabases({
connectionId: 'prod-metabase',
discovered: [
{ id: 1, name: 'Warehouse A', engine: 'postgres', host: 'localhost', dbName: 'a' },
{ id: 2, name: 'Warehouse B', engine: 'postgres', host: 'localhost', dbName: 'b' },
],
});
}
@ -151,22 +137,10 @@ describe('runLocalMetabaseIngest', () => {
});
it('throws before runner work when there are no sync-enabled mapped rows', async () => {
const store = new LocalMetabaseSourceStateReader({ dbPath: join(tempDir, '.ktx', 'db.sqlite') });
await store.replaceSourceState({
connectionId: 'prod-metabase',
mappings: [
{
metabaseDatabaseId: 1,
metabaseDatabaseName: 'Warehouse A',
metabaseEngine: 'postgres',
metabaseHost: null,
metabaseDbName: null,
targetConnectionId: null,
syncEnabled: true,
source: 'refresh',
},
],
});
project.config.connections['prod-metabase'].mappings = {
databaseMappings: { '1': null },
syncEnabled: { '1': true },
};
await expect(
runLocalMetabaseIngest({
@ -178,59 +152,28 @@ describe('runLocalMetabaseIngest', () => {
).rejects.toThrow('no sync-enabled mappings with a target connection');
});
it('throws with refresh guidance for unhydrated sync-enabled rows', async () => {
const store = new LocalMetabaseSourceStateReader({ dbPath: join(tempDir, '.ktx', 'db.sqlite') });
await store.replaceSourceState({
connectionId: 'prod-metabase',
mappings: [
{
metabaseDatabaseId: 7,
metabaseDatabaseName: null,
metabaseEngine: null,
metabaseHost: null,
metabaseDbName: null,
targetConnectionId: 'warehouse_a',
syncEnabled: true,
source: 'ktx.yaml',
},
],
it('seeds yaml-only Metabase mappings before the unhydrated fan-out preflight', async () => {
project.config.connections['prod-metabase'].mappings = {
databaseMappings: { '1': 'warehouse_a' },
syncEnabled: { '1': true },
};
const result = await runLocalMetabaseIngest({
project,
adapters: [new FakeMetabaseSourceAdapter()],
metabaseConnectionId: 'prod-metabase',
agentRunner: new TestAgentRunner(),
jobIdFactory: () => 'metabase-child-1',
});
await expect(
runLocalMetabaseIngest({
project,
adapters: [new FakeMetabaseSourceAdapter()],
expect(result.status).toBe('all_succeeded');
expect(result.children).toMatchObject([
{
metabaseConnectionId: 'prod-metabase',
agentRunner: new TestAgentRunner(),
}),
).rejects.toThrow('run `ktx connection mapping refresh prod-metabase`');
});
it('seeds yaml-only Metabase mappings before the unhydrated fan-out preflight', async () => {
const project = {
projectDir: tempDir,
config: {
ingest: { adapters: ['metabase'] },
connections: {
'prod-metabase': {
driver: 'metabase',
mappings: {
databaseMappings: { '1': 'prod-warehouse' },
syncEnabled: { '1': true },
},
},
'prod-warehouse': { driver: 'postgres', url: 'postgresql://readonly@db.test/analytics' },
},
metabaseDatabaseId: 1,
targetConnectionId: 'warehouse_a',
},
} as never;
await expect(
runLocalMetabaseIngest({
project,
adapters: [new FakeMetabaseSourceAdapter()],
metabaseConnectionId: 'prod-metabase',
}),
).rejects.toThrow('run `ktx connection mapping refresh prod-metabase`');
]);
});
it('rejects source-dir uploads through the Metabase fan-out runner', async () => {
@ -266,15 +209,15 @@ describe('runLocalMetabaseIngest', () => {
it('captures fetch-time child failures and continues later mappings', async () => {
await seedMetabaseState();
project.config.connections.warehouse_c = { driver: 'postgres', url: 'postgres://localhost/c' };
const store = new LocalMetabaseSourceStateReader({ dbPath: join(tempDir, '.ktx', 'db.sqlite') });
await store.upsertDatabaseMapping({
connectionId: 'prod-metabase',
metabaseDatabaseId: 3,
targetConnectionId: 'warehouse_c',
syncEnabled: true,
source: 'cli',
});
await store.refreshDiscoveredDatabases({
project.config.connections['prod-metabase'].mappings = {
databaseMappings: { '1': 'warehouse_a', '2': 'warehouse_b', '3': 'warehouse_c' },
syncEnabled: { '1': true, '2': true, '3': true },
syncMode: 'ALL',
defaultTagNames: ['ktx'],
selections: { collections: [], items: [] },
};
const discoveryCache = new LocalMetabaseDiscoveryCache({ dbPath: join(tempDir, '.ktx', 'db.sqlite') });
await discoveryCache.refreshDiscoveredDatabases({
connectionId: 'prod-metabase',
discovered: [
{ id: 1, name: 'Warehouse A', engine: 'postgres', host: 'localhost', dbName: 'a' },

4
packages/context/src/package-exports.test.ts
View file

@ -199,7 +199,9 @@ describe('@ktx/context package exports', () => {
expect(ingest.stagedSyncConfigSchema).toBeDefined();
expect(ingest.stagedLookerScopeFileSchema).toBeDefined();
expect(ingest.stagedLookerFetchReportSchema).toBeDefined();
expect(ingest.LocalMetabaseSourceStateReader).toBeTypeOf('function');
expect('LocalMetabaseSourceStateReader' in ingest).toBe(false);
expect(ingest.KtxYamlMetabaseSourceStateReader).toBeTypeOf('function');
expect(ingest.LocalMetabaseDiscoveryCache).toBeTypeOf('function');
expect(ingest.createLocalMetabaseSourceAdapter).toBeTypeOf('function');
expect(ingest.metabaseRuntimeConfigFromLocalConnection).toBeTypeOf('function');
expect(ingest.IngestMetabaseClientFactory).toBeTypeOf('function');

16
packages/context/src/project/config.ts
View file

@ -505,13 +505,15 @@ export function parseKtxProjectConfig(raw: string): KtxProjectConfig {
return {
project: project.trim(),
...(setup
? {
setup: {
database_connection_ids: stringArray(setup.database_connection_ids, []),
completed_steps: stringArray(setup.completed_steps, []),
},
}
: {}),
? {
setup: {
database_connection_ids: stringArray(setup.database_connection_ids, []),
...(setup.completed_steps !== undefined
? { completed_steps: stringArray(setup.completed_steps, []) }
: {}),
},
}
: {}),
connections: isRecord(parsed.connections)
? (parsed.connections as Record<string, KtxProjectConnectionConfig>)
: defaults.connections,

1
packages/context/src/project/setup-config.ts
View file

@ -95,7 +95,6 @@ export function setKtxSetupDatabaseConnectionIds(
...config,
setup: {
database_connection_ids: uniqueConnectionIds,
...(config.setup?.completed_steps ? { completed_steps: [...config.setup.completed_steps] } : {}),
},
};
}