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

Fix historic SQL ingest setup and progress

Browse source
This commit is contained in:
Andrey Avtomonov 2026-05-11 22:35:07 +02:00
parent f3f6b36551
commit 1bd29c7eb1
14 changed files with 877 additions and 34 deletions
Download patch file Download diff file Expand all files Collapse all files

19
packages/connector-postgres/src/connector.test.ts
View file

@ -129,6 +129,25 @@ describe('KtxPostgresScanConnector', () => {
options: '-c search_path=analytics,public',
ssl: { rejectUnauthorized: false },
});
const libpqPreferConfig = postgresPoolConfigFromConfig({
connectionId: 'warehouse',
connection: {
driver: 'postgres',
url: 'env:DEMO_DATABASE_URL',
readonly: true,
},
env: {
DEMO_DATABASE_URL: 'postgresql://reader@start.kaelio.com:5432/demo?sslmode=prefer',
},
});
expect(libpqPreferConfig).toMatchObject({
host: 'start.kaelio.com',
port: 5432,
database: 'demo',
user: 'reader',
});
expect(libpqPreferConfig).not.toHaveProperty('connectionString');
expect(libpqPreferConfig).not.toHaveProperty('ssl');
expect(() =>
postgresPoolConfigFromConfig({
connectionId: 'warehouse',

14
packages/connector-postgres/src/connector.ts
View file

@ -57,6 +57,8 @@ export interface KtxPostgresConnectionConfig {
schema?: string;
schemas?: string[];
ssl?: boolean;
sslmode?: string;
sslMode?: string;
rejectUnauthorized?: boolean;
readonly?: boolean;
[key: string]: unknown;
@ -253,15 +255,22 @@ function numberValue(value: unknown): number | undefined {
function parsePostgresUrl(url: string): Partial<KtxPostgresConnectionConfig> {
const parsed = new URL(url);
const sslmode = parsed.searchParams.get('sslmode') ?? undefined;
return {
host: parsed.hostname,
port: parsed.port ? Number(parsed.port) : undefined,
database: parsed.pathname.replace(/^\/+/, '') || undefined,
username: parsed.username ? decodeURIComponent(parsed.username) : undefined,
password: parsed.password ? decodeURIComponent(parsed.password) : undefined,
...(sslmode ? { sslmode } : {}),
};
}
function normalizedSslMode(connection: KtxPostgresConnectionConfig): string | undefined {
const value = connection.sslmode ?? connection.sslMode;
return typeof value === 'string' && value.trim().length > 0 ? value.trim().toLowerCase() : undefined;
}
function schemasFromConnection(connection: KtxPostgresConnectionConfig): string[] {
if (Array.isArray(connection.schemas) && connection.schemas.length > 0) {
return connection.schemas.filter((schema): schema is string => typeof schema === 'string' && schema.length > 0);
@ -299,6 +308,7 @@ export function postgresPoolConfigFromConfig(input: {
const database = stringConfigValue(merged, 'database', env);
const user = stringConfigValue(merged, 'username', env) ?? stringConfigValue(merged, 'user', env);
const password = stringConfigValue(merged, 'password', env);
const sslmode = normalizedSslMode(merged);
if (!referencedUrl && !host) {
throw new Error(`Native PostgreSQL connector requires connections.${input.connectionId}.host or url`);
@ -314,7 +324,7 @@ export function postgresPoolConfigFromConfig(input: {
max: 10,
idleTimeoutMillis: 30_000,
connectionTimeoutMillis: 10_000,
...(referencedUrl
...(referencedUrl && sslmode !== 'prefer' && sslmode !== 'disable'
? { connectionString: referencedUrl }
: { host, port: numberValue(merged.port) ?? 5432, database, user, password }),
};
@ -322,7 +332,7 @@ export function postgresPoolConfigFromConfig(input: {
if (searchPathSchemas.length > 0) {
config.options = `-c search_path=${searchPathSchemas.join(',')}`;
}
if (merged.ssl) {
if (merged.ssl && sslmode !== 'prefer' && sslmode !== 'disable') {
config.ssl = { rejectUnauthorized: merged.rejectUnauthorized ?? true };
}
return config;