77 lines
2.7 KiB
TypeScript
77 lines
2.7 KiB
TypeScript
import { randomUUID } from 'node:crypto';
|
|
import { afterAll, beforeAll, describe, expect, it } from 'vitest';
|
|
import {
|
|
connectorLeaseAuditLog,
|
|
createDb,
|
|
eq,
|
|
logicalAgentConnectorLeases,
|
|
type DbHandle,
|
|
} from '@mosaicstack/db';
|
|
import { ConnectorLeaseCoordinator } from '@mosaicstack/agent';
|
|
import { ConnectorLeaseRepository } from './connector-lease.repository.js';
|
|
|
|
const hasPostgres = Boolean(process.env['DATABASE_URL']);
|
|
const tenantId = `lease-test-${randomUUID()}`;
|
|
const identity = { tenantId, logicalAgentId: 'mos' } as const;
|
|
|
|
describe.skipIf(!hasPostgres)('ConnectorLeaseRepository real PostgreSQL integration', (): void => {
|
|
let handle: DbHandle;
|
|
|
|
beforeAll((): void => {
|
|
handle = createDb(process.env['DATABASE_URL']);
|
|
});
|
|
|
|
afterAll(async (): Promise<void> => {
|
|
if (!handle) return;
|
|
await handle.db
|
|
.delete(connectorLeaseAuditLog)
|
|
.where(eq(connectorLeaseAuditLog.tenantId, tenantId));
|
|
await handle.db
|
|
.delete(logicalAgentConnectorLeases)
|
|
.where(eq(logicalAgentConnectorLeases.tenantId, tenantId));
|
|
await handle.close();
|
|
});
|
|
|
|
it('preserves the exclusive CAS fence across a real pool close/reopen', async (): Promise<void> => {
|
|
const command = {
|
|
identity,
|
|
bindingId: 'operator-chat',
|
|
scopes: ['runtime.send'],
|
|
ttlMs: 60_000,
|
|
} as const;
|
|
const firstCoordinator = new ConnectorLeaseCoordinator(new ConnectorLeaseRepository(handle.db));
|
|
const contenders = await Promise.allSettled([
|
|
firstCoordinator.acquire({
|
|
...command,
|
|
connectorId: 'connector-a',
|
|
correlationId: 'postgres-acquire-a',
|
|
}),
|
|
firstCoordinator.acquire({
|
|
...command,
|
|
connectorId: 'connector-b',
|
|
correlationId: 'postgres-acquire-b',
|
|
}),
|
|
]);
|
|
const acquired = contenders.find((result) => result.status === 'fulfilled');
|
|
if (!acquired || acquired.status !== 'fulfilled') throw new Error('no lease contender won');
|
|
expect(contenders.filter((result) => result.status === 'fulfilled')).toHaveLength(1);
|
|
|
|
await handle.close();
|
|
handle = createDb(process.env['DATABASE_URL']);
|
|
const reopened = new ConnectorLeaseCoordinator(new ConnectorLeaseRepository(handle.db));
|
|
const persisted = await reopened.current({ identity, bindingId: 'operator-chat' });
|
|
expect(persisted).toMatchObject({
|
|
leaseId: acquired.value.leaseId,
|
|
leaseEpoch: '1',
|
|
});
|
|
|
|
const takeover = await reopened.takeover({
|
|
...command,
|
|
connectorId: 'connector-c',
|
|
correlationId: 'postgres-takeover',
|
|
expectedEpoch: acquired.value.leaseEpoch,
|
|
});
|
|
expect(takeover).toMatchObject({ connectorId: 'connector-c', leaseEpoch: '2' });
|
|
});
|
|
});
|