This repository has no description
Something went wrong. Try again.
8.3 kB · 249 lines
TypeScript
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250// SPDX-License-Identifier: AGPL-3.0-or-later
import {randomUUID} from 'node:crypto';import {existsSync} from 'node:fs';import {getDefaultPostgresClient, initPostgres, quoteIdentifier, shutdownPostgres} from '@pkgs/postgres/src/Client';import cassandra from 'cassandra-driver';import {afterAll, beforeAll, describe, expect, test} from 'vitest';import {defineTable} from './CassandraTableDsl';import {Db} from './CassandraTypes';import {ensurePostgresKvSchema, PostgresKvQueryExecutor, pruneExpiredPostgresKvRows} from './PostgresKvQueryExecutor';
interface TestRow { tenant_id: string; item_id: bigint; created_at: Date; payload: Buffer | null; tags: Set<string> | null; counts: Map<string, bigint> | null; local_day: cassandra.types.LocalDate | null; note: string | null;}
function postgresHost(): string { return process.env.FLUXER_POSTGRES_TEST_HOST ?? (existsSync('/.dockerenv') ? 'host.docker.internal' : '127.0.0.1');}
function sleep(ms: number): Promise<void> { return new Promise((resolve) => setTimeout(resolve, ms));}
const kvTable = `fluxer_kv_test_${randomUUID().replaceAll('-', '_')}`;const logicalTableName = `pg_kv_test_${randomUUID().replaceAll('-', '_')}`;const TestRows = defineTable<TestRow, 'tenant_id' | 'item_id', 'tenant_id'>({ name: logicalTableName, columns: ['tenant_id', 'item_id', 'created_at', 'payload', 'tags', 'counts', 'local_day', 'note'], primaryKey: ['tenant_id', 'item_id'], partitionKey: ['tenant_id'],});
describe('PostgresKvQueryExecutor', () => { let executor: PostgresKvQueryExecutor; let initialized = false;
beforeAll(async () => { await initPostgres({ host: postgresHost(), port: Number(process.env.FLUXER_POSTGRES_PORT ?? 5432), database: process.env.FLUXER_POSTGRES_DATABASE ?? 'fluxer', username: process.env.FLUXER_POSTGRES_USERNAME ?? 'fluxer', password: process.env.FLUXER_POSTGRES_PASSWORD ?? 'fluxer', kvTable, maxConnections: 4, }); initialized = true; const client = getDefaultPostgresClient(); await ensurePostgresKvSchema(client); executor = new PostgresKvQueryExecutor(client); });
afterAll(async () => { if (!initialized) return; try { await getDefaultPostgresClient().query(`DROP TABLE IF EXISTS ${quoteIdentifier(kvTable)}`); } finally { await shutdownPostgres(); } });
test('round trips typed values and conditional writes through a real Postgres KV table', async () => { const row: TestRow = { tenant_id: 'tenant-a', item_id: 10n, created_at: new Date('2026-01-02T03:04:05.006Z'), payload: Buffer.from('hello postgres kv'), tags: new Set(['alpha', 'beta']), counts: new Map([['seen', 5n]]), local_day: cassandra.types.LocalDate.fromString('2026-01-02'), note: 'first', };
await expect(executor.executeQuery<{['[applied]']: boolean}>(TestRows.insertIfNotExists(row))).resolves.toEqual([ {'[applied]': true}, ]); await expect( executor.executeQuery<{['[applied]']: boolean}>(TestRows.insertIfNotExists({...row, note: 'second'})), ).resolves.toEqual([{'[applied]': false}]);
const fetched = await executor.executeQuery<TestRow>( TestRows.select({ where: [TestRows.where.eq('tenant_id'), TestRows.where.eq('item_id')], limit: 1, }).bind({tenant_id: 'tenant-a', item_id: 10n}), );
expect(fetched).toHaveLength(1); expect(fetched[0]!.item_id).toBe(10n); expect(fetched[0]!.created_at.getTime()).toBe(row.created_at.getTime()); expect(fetched[0]!.payload?.equals(row.payload!)).toBe(true); expect([...fetched[0]!.tags!.values()].sort()).toEqual(['alpha', 'beta']); expect(fetched[0]!.counts!.get('seen')).toBe(5n); expect(fetched[0]!.local_day?.toString()).toBe('2026-01-02'); expect(fetched[0]!.note).toBe('first'); });
test('filters, orders, patches, pages, and deletes logical rows', async () => { await executor.executeBatch( [ TestRows.upsertAll({ tenant_id: 'tenant-b', item_id: 1n, created_at: new Date('2026-02-01T00:00:00.000Z'), payload: null, tags: null, counts: null, local_day: null, note: 'one', }), TestRows.upsertAll({ tenant_id: 'tenant-b', item_id: 2n, created_at: new Date('2026-02-02T00:00:00.000Z'), payload: null, tags: null, counts: null, local_day: null, note: 'two', }), TestRows.upsertAll({ tenant_id: 'tenant-b', item_id: 3n, created_at: new Date('2026-02-03T00:00:00.000Z'), payload: null, tags: null, counts: null, local_day: null, note: 'three', }), ].map((query) => ({query: query.cql, params: query.params, meta: query.kvMeta})), );
const firstPage = await executor.executePagedQuery<TestRow>( TestRows.select({ where: TestRows.where.eq('tenant_id'), orderBy: {col: 'item_id', direction: 'DESC'}, }).bind({tenant_id: 'tenant-b'}), {pageSize: 2}, ); expect(firstPage.rows.map((row) => row.item_id)).toEqual([3n, 2n]); expect(firstPage.pageState).not.toBeNull();
const secondPage = await executor.executePagedQuery<TestRow>( TestRows.select({ where: TestRows.where.eq('tenant_id'), orderBy: {col: 'item_id', direction: 'DESC'}, }).bind({tenant_id: 'tenant-b'}), {pageSize: 2, pageState: firstPage.pageState}, ); expect(secondPage.rows.map((row) => row.item_id)).toEqual([1n]); expect(secondPage.pageState).toBeNull();
await executor.executeQuery(TestRows.patchByPk({tenant_id: 'tenant-b', item_id: 2n}, {note: Db.clear()})); const patched = await executor.executeQuery<TestRow>( TestRows.select({where: [TestRows.where.eq('tenant_id'), TestRows.where.eq('item_id')], limit: 1}).bind({ tenant_id: 'tenant-b', item_id: 2n, }), ); expect(patched[0]!.note).toBeNull();
await executor.executeQuery(TestRows.delete({where: TestRows.where.eq('tenant_id')}).bind({tenant_id: 'tenant-b'})); const remaining = await executor.executeQuery<TestRow>( TestRows.select({where: TestRows.where.eq('tenant_id')}).bind({tenant_id: 'tenant-b'}), ); expect(remaining).toEqual([]); });
test('filters expired rows and prunes them physically', async () => { const client = getDefaultPostgresClient(); await executor.executeQuery( TestRows.upsertAllWithTtl( { tenant_id: 'tenant-expired', item_id: 1n, created_at: new Date('2026-03-01T00:00:00.000Z'), payload: null, tags: null, counts: null, local_day: null, note: 'temporary', }, 1, ), );
await sleep(1100);
const visibleRows = await executor.executeQuery<TestRow>( TestRows.select({where: TestRows.where.eq('tenant_id')}).bind({tenant_id: 'tenant-expired'}), ); expect(visibleRows).toEqual([]);
const physicalBefore = await client.query<{count: string}>( `SELECT count(*)::text AS count FROM ${quoteIdentifier(kvTable)} WHERE table_name = $1 AND row_data ->> 'tenant_id' = $2`, [logicalTableName, 'tenant-expired'], ); expect(Number(physicalBefore.rows[0]!.count)).toBe(1);
const pruned = await pruneExpiredPostgresKvRows(client, 10); expect(pruned).toBeGreaterThanOrEqual(1);
const physicalAfter = await client.query<{count: string}>( `SELECT count(*)::text AS count FROM ${quoteIdentifier(kvTable)} WHERE table_name = $1 AND row_data ->> 'tenant_id' = $2`, [logicalTableName, 'tenant-expired'], ); expect(Number(physicalAfter.rows[0]!.count)).toBe(0); });
test('patching a logically expired row makes it visible again without keeping stale TTL', async () => { await executor.executeQuery( TestRows.upsertAllWithTtl( { tenant_id: 'tenant-resurrect', item_id: 1n, created_at: new Date('2026-04-01T00:00:00.000Z'), payload: null, tags: null, counts: null, local_day: null, note: 'temporary', }, 1, ), );
await sleep(1100); await executor.executeQuery( TestRows.patchByPk({tenant_id: 'tenant-resurrect', item_id: 1n}, {note: Db.set('resurrected')}), );
const rows = await executor.executeQuery<TestRow>( TestRows.select({ where: [TestRows.where.eq('tenant_id'), TestRows.where.eq('item_id')], limit: 1, }).bind({tenant_id: 'tenant-resurrect', item_id: 1n}), ); expect(rows).toHaveLength(1); expect(rows[0]!.note).toBe('resurrected'); });});