Function main

packages/db/scripts/migrate-sessions.ts:43–116

()

const PROJECTS_SET_KEY = 'session:projects';

async function main() {
  const dryRun = process.argv.includes('--dry-run');
  const redis = getRedisCache();

  console.log(
    `[migrate-sessions] starting${dryRun ? ' (DRY RUN — no writes)' : ''}`
  );

  const jobs = await sessionsQueue.getJobs(['delayed', 'waiting']);
  console.log(`[migrate-sessions] found ${jobs.length} legacy session jobs`);

  const counts = {
    migrated: 0,
    blobMissing: 0,
    alreadyMigrated: 0,
    parseFailed: 0,
  };

  for (const job of jobs) {
    const data = job.data as EventsQueuePayloadCreateSessionEnd | undefined;
    const payload = data?.payload;
    if (!payload?.projectId || !payload.deviceId || !payload.sessionId) {
      counts.parseFailed++;
      continue;
    }
    const { projectId, deviceId, sessionId } = payload;

    const existsAlready = await redis.exists(sessionKey(projectId, deviceId));
    if (existsAlready) {
      counts.alreadyMigrated++;
      continue;
    }

    const blob = await redis.get(`session:${sessionId}`);
    if (!blob) {
      counts.blobMissing++;
      continue;
    }

    const session = getSafeJson<IClickhouseSession>(blob);
    if (!session) {
      counts.parseFailed++;
      continue;
    }

    const nowMs = Date.now();

    if (dryRun) {
      counts.migrated++;
      continue;
    }

    const multi = redis.multi();
    multi.set(sessionKey(projectId, deviceId), blob);
    multi.zadd(wallclockSetKey(projectId), nowMs.toString(), deviceId);
    multi.sadd(PROJECTS_SET_KEY, projectId);
    if (session.profile_id && session.profile_id !== deviceId) {
      multi.set(profileIndexKey(projectId, session.profile_id), deviceId);