Merge pull request #1582 from dcoric/feat/postgres-data-migration

feat: data migration from mongo/fs to the PostgreSQL sink

Author: dcoric

package.json

@@ -54,6 +54,7 @@
   "scripts": {
     "cli": "tsx ./packages/git-proxy-cli/index.ts",
     "cli:js": "node ./packages/git-proxy-cli/dist/index.js",
+    "migrate:postgres": "tsx scripts/migrate-to-postgres.ts",
     "client": "vite --config vite.config.ts",
     "clientinstall": "npm install --prefix client",
     "server": "cross-env ALLOWED_ORIGINS=* tsx index.ts",

scripts/migrate-to-postgres.ts

@@ -0,0 +1,87 @@
+#!/usr/bin/env tsx
+
+/**
+ * Copyright 2026 GitProxy Contributors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import yargs from 'yargs';
+import { hideBin } from 'yargs/helpers';
+
+import { getDatabase } from '../src/config';
+import * as postgres from '../src/db/postgres';
+import { resetConnection } from '../src/db/postgres/helper';
+import { migrate, MigrationSource } from '../src/db/postgres/migrate';
+import { createFileSource } from '../src/db/postgres/migrateFileSource';
+import { createMongoSource } from '../src/db/postgres/migrateMongoSource';
+
+const argv = yargs(hideBin(process.argv))
+  .usage('Usage: $0 --from <mongo|fs> [options]')
+  .option('from', {
+    choices: ['mongo', 'fs'] as const,
+    demandOption: true,
+    describe: 'Source backend to migrate from',
+  })
+  .option('mongoUrl', {
+    type: 'string',
+    describe: 'MongoDB connection string (required when --from mongo)',
+  })
+  .option('dataDir', {
+    type: 'string',
+    describe: 'NeDB data directory (defaults to ./.data/db) when --from fs',
+  })
+  .strict()
+  .parseSync();
+
+const buildSource = async (): Promise<MigrationSource> => {
+  if (argv.from === 'mongo') {
+    if (!argv.mongoUrl) {
+      throw new Error('--mongoUrl is required when --from mongo');
+    }
+    return createMongoSource(argv.mongoUrl);
+  }
+  return createFileSource(argv.dataDir);
+};
+
+const main = async (): Promise<void> => {
+  // The destination is the active sink, so it must be postgres. Reading the
+  // source is independent (its own driver), so the two never clash.
+  const db = getDatabase();
+  if (db.type !== 'postgres') {
+    throw new Error(
+      `The active sink is "${db.type}", but this migration writes to postgres. ` +
+        'Enable the postgres sink (with its connectionString or ' +
+        'GIT_PROXY_POSTGRES_CONNECTION_STRING) before running this.',
+    );
+  }
+
+  const source = await buildSource();
+  try {
+    const summary = await migrate(source, postgres, { log: (message) => console.log(message) });
+    console.log('Migration complete:');
+    console.log(`  users:  ${summary.users.imported} imported, ${summary.users.skipped} skipped`);
+    console.log(`  repos:  ${summary.repos.imported} imported, ${summary.repos.skipped} skipped`);
+    console.log(`  pushes: ${summary.pushes.imported} imported`);
+  } finally {
+    await source.close();
+    // Close the destination pool too, or its open handles keep the process
+    // alive after the summary prints.
+    await resetConnection();
+  }
+};
+
+main().catch((err) => {
+  console.error(`Migration failed: ${err instanceof Error ? err.message : err}`);
+  process.exit(1);
+});

src/db/postgres/migrate.ts

@@ -0,0 +1,127 @@
+/**
+ * Copyright 2026 GitProxy Contributors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import { Action } from '../../proxy/actions';
+import { Repo, User } from '../types';
+
+/**
+ * A read-only view over a backend (mongo or fs) that data is migrated *from*.
+ * Implementations own their own connection and must be closed by the caller.
+ */
+export const DEFAULT_PUSH_BATCH_SIZE = 500;
+
+export interface MigrationSource {
+  getUsers(): Promise<User[]>;
+  getRepos(): Promise<Repo[]>;
+  /**
+   * Stream pushes in batches of at most `batchSize`. Production datasets can
+   * hold tens of thousands of large push documents, so sources must not
+   * require the whole table in memory at once.
+   */
+  getPushBatches(batchSize: number): AsyncIterable<Action[]>;
+  close(): Promise<void>;
+}
+
+/**
+ * The subset of the Postgres adapter used to write migrated records. The
+ * adapter module satisfies this shape directly, so the CLI can pass it as-is.
+ */
+export interface MigrationDestination {
+  findUser(username: string): Promise<User | null>;
+  findUserByEmail(email: string): Promise<User | null>;
+  createUser(user: User): Promise<void>;
+  getRepoByUrl(url: string): Promise<Repo | null>;
+  createRepo(repo: Repo): Promise<Repo>;
+  writeAudit(action: Action): Promise<void>;
+}
+
+export interface MigrationSummary {
+  users: { imported: number; skipped: number };
+  repos: { imported: number; skipped: number };
+  pushes: { imported: number };
+}
+
+export interface MigrateOptions {
+  /** Receives human-readable progress lines. Defaults to a no-op. */
+  log?: (message: string) => void;
+  /** Maximum pushes fetched and written per batch. Defaults to 500. */
+  pushBatchSize?: number;
+}
+
+/**
+ * Copy users, repos and pushes from `source` into the Postgres `destination`.
+ *
+ * Idempotent and re-runnable: users and repos that already exist (matched by
+ * username/email and URL respectively) are skipped, and pushes are upserted by
+ * their stable string id. Record `_id`s are intentionally NOT carried over —
+ * Postgres assigns fresh UUIDs; push ids (TEXT) are preserved by the upsert.
+ */
+export const migrate = async (
+  source: MigrationSource,
+  destination: MigrationDestination,
+  options: MigrateOptions = {},
+): Promise<MigrationSummary> => {
+  const log = options.log ?? (() => undefined);
+  const summary: MigrationSummary = {
+    users: { imported: 0, skipped: 0 },
+    repos: { imported: 0, skipped: 0 },
+    pushes: { imported: 0 },
+  };
+
+  const users = await source.getUsers();
+  log(`Migrating ${users.length} user(s)...`);
+  for (const user of users) {
+    const existing =
+      (await destination.findUser(user.username)) ||
+      (user.email ? await destination.findUserByEmail(user.email) : null);
+    if (existing) {
+      summary.users.skipped++;
+      continue;
+    }
+    // Legacy documents can lack optional fields the writers dereference:
+    // users synced from AD may have no email (the mail attribute is
+    // optional) or gitAccount. Default them like the mongo upsert path does.
+    await destination.createUser({
+      ...user,
+      email: user.email ?? '',
+      gitAccount: user.gitAccount ?? '',
+    });
+    summary.users.imported++;
+  }
+
+  const repos = await source.getRepos();
+  log(`Migrating ${repos.length} repo(s)...`);
+  for (const repo of repos) {
+    if (await destination.getRepoByUrl(repo.url)) {
+      summary.repos.skipped++;
+      continue;
+    }
+    await destination.createRepo(repo);
+    summary.repos.imported++;
+  }
+
+  const batchSize = options.pushBatchSize ?? DEFAULT_PUSH_BATCH_SIZE;
+  log('Migrating pushes...');
+  for await (const batch of source.getPushBatches(batchSize)) {
+    for (const push of batch) {
+      await destination.writeAudit(push);
+      summary.pushes.imported++;
+    }
+    log(`  ${summary.pushes.imported} push(es) migrated`);
+  }
+
+  return summary;
+};

src/db/postgres/migrateFileSource.ts

@@ -0,0 +1,97 @@
+/**
+ * Copyright 2026 GitProxy Contributors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import fs from 'fs';
+import path from 'path';
+
+import Datastore from '@seald-io/nedb';
+
+import { Action } from '../../proxy/actions';
+import { toClass } from '../helper';
+import { Repo, User } from '../types';
+import { MigrationSource } from './migrate';
+
+// Where the `fs` sink keeps its NeDB datastores.
+const DEFAULT_DATA_DIR = './.data/db';
+
+/**
+ * Build a read-only {@link MigrationSource} backed by the NeDB datastores the
+ * `fs` sink writes. `dataDir` defaults to the location the sink uses
+ * (`./.data/db`). Record `_id`s are ignored by the Postgres writers, which
+ * assign fresh UUIDs.
+ */
+const DATASTORE_FILES = ['users.db', 'repos.db', 'pushes.db'];
+
+interface LazyStore {
+  store: Datastore;
+  ready: () => Promise<void>;
+}
+
+export const createFileSource = (dataDir: string = DEFAULT_DATA_DIR): MigrationSource => {
+  // Fail fast on a wrong path rather than reporting a legitimately empty
+  // backend: a missing directory, or one containing none of the fs sink's
+  // datastores, is a misconfiguration, while an existing-but-empty datastore
+  // is a real (empty) backend.
+  if (!fs.existsSync(dataDir)) {
+    throw new Error(`fs sink data directory does not exist: ${dataDir}`);
+  }
+  if (!DATASTORE_FILES.some((file) => fs.existsSync(path.join(dataDir, file)))) {
+    throw new Error(`No fs sink datastores (${DATASTORE_FILES.join(', ')}) found in: ${dataDir}`);
+  }
+
+  // Loading is explicit (no autoload) so a corrupt datastore surfaces as a
+  // clear error instead of being silently treated as empty.
+  const load = (file: string): LazyStore => {
+    const filename = path.join(dataDir, file);
+    const store = new Datastore({ filename });
+    let loading: Promise<void> | undefined;
+    const ready = () =>
+      (loading ??= store.loadDatabaseAsync().catch((err: unknown) => {
+        throw new Error(
+          `Failed to load ${filename}: ${err instanceof Error ? err.message : String(err)}`,
+        );
+      }));
+    return { store, ready };
+  };
+
+  const users = load('users.db');
+  const repos = load('repos.db');
+  const pushes = load('pushes.db');
+
+  const readAll = async <T>({ store, ready }: LazyStore, proto: object): Promise<T[]> => {
+    await ready();
+    const docs = await store.findAsync<Record<string, unknown>>({});
+    return docs.map((doc) => toClass(doc, proto) as T);
+  };
+
+  // NeDB keeps the whole datastore in memory regardless, so batching here only
+  // shapes the write side to match the MigrationSource contract.
+  const getPushBatches = (batchSize: number): AsyncIterable<Action[]> => ({
+    async *[Symbol.asyncIterator]() {
+      const all = await readAll<Action>(pushes, Action.prototype);
+      for (let i = 0; i < all.length; i += batchSize) {
+        yield all.slice(i, i + batchSize);
+      }
+    },
+  });
+
+  return {
+    getUsers: () => readAll<User>(users, User.prototype),
+    getRepos: () => readAll<Repo>(repos, Repo.prototype),
+    getPushBatches,
+    close: () => Promise.resolve(),
+  };
+};

src/db/postgres/migrateMongoSource.ts

@@ -0,0 +1,72 @@
+/**
+ * Copyright 2026 GitProxy Contributors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import { MongoClient, MongoClientOptions } from 'mongodb';
+
+import { Action } from '../../proxy/actions';
+import { toClass } from '../helper';
+import { Repo, User } from '../types';
+import { MigrationSource } from './migrate';
+
+/**
+ * Build a read-only {@link MigrationSource} backed by a MongoDB instance.
+ *
+ * The connection is explicit (not the configured sink) so the importer can read
+ * the legacy backend while the active sink points at the Postgres destination.
+ * The caller owns the lifecycle and must `close()` the source when done.
+ */
+export const createMongoSource = async (
+  connectionString: string,
+  options: MongoClientOptions = {},
+): Promise<MigrationSource> => {
+  const client = new MongoClient(connectionString, options);
+  await client.connect();
+  const db = client.db();
+
+  const readAll = async <T>(collection: string, proto: object): Promise<T[]> => {
+    const docs = await db.collection(collection).find().toArray();
+    // toClass drops mongo class identity; the ObjectId `_id` it carries through
+    // is ignored by the Postgres writers, which assign fresh UUIDs.
+    return docs.map((doc) => toClass(doc, proto) as T);
+  };
+
+  // Pushes are streamed through a cursor rather than materialised: production
+  // tables can hold tens of thousands of documents that each carry a full
+  // diff, so a single toArray() would hold the entire table in memory.
+  const getPushBatches = (batchSize: number): AsyncIterable<Action[]> => ({
+    async *[Symbol.asyncIterator]() {
+      const cursor = db.collection('pushes').find().batchSize(batchSize);
+      let batch: Action[] = [];
+      for await (const doc of cursor) {
+        batch.push(toClass(doc, Action.prototype) as Action);
+        if (batch.length >= batchSize) {
+          yield batch;
+          batch = [];
+        }
+      }
+      if (batch.length > 0) {
+        yield batch;
+      }
+    },
+  });
+
+  return {
+    getUsers: () => readAll<User>('users', User.prototype),
+    getRepos: () => readAll<Repo>('repos', Repo.prototype),
+    getPushBatches,
+    close: () => client.close(),
+  };
+};