stats/packages/queue/src/queues.ts

import { createHash } from 'node:crypto';
import type {
  IServiceCreateEventPayload,
  IServiceEvent,
  Prisma,
} from '@openpanel/db';
import { createLogger } from '@openpanel/logger';
import { getRedisGroupQueue, getRedisQueue } from '@openpanel/redis';
import { Queue } from 'bullmq';
import { Queue as GroupQueue } from 'groupmq';
import type { ITrackPayload } from '../../validation';

export const EVENTS_GROUP_QUEUES_SHARDS = Number.parseInt(
  process.env.EVENTS_GROUP_QUEUES_SHARDS || '1',
  10
);

export const getQueueName = (name: string) =>
  process.env.QUEUE_CLUSTER ? `{${name}}` : name;

function pickShard(projectId: string) {
  const h = createHash('sha1').update(projectId).digest(); // 20 bytes
  // take first 4 bytes as unsigned int
  const x = h.readUInt32BE(0);
  return x % EVENTS_GROUP_QUEUES_SHARDS; // 0..n-1
}

export const queueLogger = createLogger({ name: 'queue' });

export interface EventsQueuePayloadIncomingEvent {
  type: 'incomingEvent';
  payload: {
    projectId: string;
    event: ITrackPayload & {
      timestamp: string | number;
      isTimestampFromThePast: boolean;
    };
    uaInfo:
      | {
          readonly isServer: true;
          readonly device: 'server';
          readonly os: '';
          readonly osVersion: '';
          readonly browser: '';
          readonly browserVersion: '';
          readonly brand: '';
          readonly model: '';
        }
      | {
          readonly os: string | undefined;
          readonly osVersion: string | undefined;
          readonly browser: string | undefined;
          readonly browserVersion: string | undefined;
          readonly device: string;
          readonly brand: string | undefined;
          readonly model: string | undefined;
          readonly isServer: false;
        };
    geo: {
      country: string | undefined;
      city: string | undefined;
      region: string | undefined;
      longitude: number | undefined;
      latitude: number | undefined;
    };
    headers: Record<string, string | undefined>;
    deviceId: string;
    sessionId: string;
    session?: Pick<
      IServiceCreateEventPayload,
      'referrer' | 'referrerName' | 'referrerType'
    >;
  };
}
export interface EventsQueuePayloadCreateEvent {
  type: 'createEvent';
  payload: Omit<IServiceEvent, 'id'>;
}

export interface EventsQueuePayloadCreateSessionEnd {
  type: 'createSessionEnd';
  payload: IServiceCreateEventPayload;
}

// TODO: Rename `EventsQueuePayloadCreateSessionEnd`
export type SessionsQueuePayload = EventsQueuePayloadCreateSessionEnd;

export type EventsQueuePayload =
  | EventsQueuePayloadCreateEvent
  | EventsQueuePayloadCreateSessionEnd
  | EventsQueuePayloadIncomingEvent;

export type CronQueuePayloadSalt = {
  type: 'salt';
  payload: undefined;
};
export type CronQueuePayloadFlushEvents = {
  type: 'flushEvents';
  payload: undefined;
};
export type CronQueuePayloadFlushProfiles = {
  type: 'flushProfiles';
  payload: undefined;
};
export type CronQueuePayloadFlushSessions = {
  type: 'flushSessions';
  payload: undefined;
};
export type CronQueuePayloadPing = {
  type: 'ping';
  payload: undefined;
};
export type CronQueuePayloadProject = {
  type: 'deleteProjects';
  payload: undefined;
};
export type CronQueuePayloadInsightsDaily = {
  type: 'insightsDaily';
  payload: undefined;
};
export type CronQueuePayloadOnboarding = {
  type: 'onboarding';
  payload: undefined;
};
export type CronQueuePayloadFlushProfileBackfill = {
  type: 'flushProfileBackfill';
  payload: undefined;
};
export type CronQueuePayloadFlushReplay = {
  type: 'flushReplay';
  payload: undefined;
};
export type CronQueuePayloadGscSync = {
  type: 'gscSync';
  payload: undefined;
};
export type CronQueuePayload =
  | CronQueuePayloadSalt
  | CronQueuePayloadFlushEvents
  | CronQueuePayloadFlushSessions
  | CronQueuePayloadFlushProfiles
  | CronQueuePayloadFlushProfileBackfill
  | CronQueuePayloadFlushReplay
  | CronQueuePayloadPing
  | CronQueuePayloadProject
  | CronQueuePayloadInsightsDaily
  | CronQueuePayloadOnboarding
  | CronQueuePayloadGscSync;

export type MiscQueuePayloadTrialEndingSoon = {
  type: 'trialEndingSoon';
  payload: {
    organizationId: string;
  };
};

export type MiscQueuePayload = MiscQueuePayloadTrialEndingSoon;

export type CronQueueType = CronQueuePayload['type'];

const orderingDelayMs = Number.parseInt(
  process.env.ORDERING_DELAY_MS || '100',
  10
);

const autoBatchMaxWaitMs = Number.parseInt(
  process.env.AUTO_BATCH_MAX_WAIT_MS || '0',
  10
);
const autoBatchSize = Number.parseInt(process.env.AUTO_BATCH_SIZE || '0', 10);

export const eventsGroupQueues = Array.from({
  length: EVENTS_GROUP_QUEUES_SHARDS,
}).map(
  (_, index, list) =>
    new GroupQueue<EventsQueuePayloadIncomingEvent['payload']>({
      logger: process.env.NODE_ENV === 'production' ? queueLogger : undefined,
      namespace: getQueueName(
        list.length === 1 ? 'group_events' : `group_events_${index}`
      ),
      redis: getRedisGroupQueue(),
      keepCompleted: 1000,
      keepFailed: 10_000,
      orderingDelayMs,
      autoBatch:
        autoBatchMaxWaitMs && autoBatchSize
          ? {
              maxWaitMs: autoBatchMaxWaitMs,
              size: autoBatchSize,
            }
          : undefined,
    })
);

export const getEventsGroupQueueShard = (groupId: string) => {
  const shard = pickShard(groupId);
  const queue = eventsGroupQueues[shard];
  if (!queue) {
    throw new Error(`Queue not found for group ${groupId}`);
  }
  return queue;
};

export const sessionsQueue = new Queue<SessionsQueuePayload>(
  getQueueName('sessions'),
  {
    connection: getRedisQueue(),
    defaultJobOptions: {
      removeOnComplete: 10,
    },
  }
);

export const cronQueue = new Queue<CronQueuePayload>(getQueueName('cron'), {
  connection: getRedisQueue(),
  defaultJobOptions: {
    removeOnComplete: 10,
  },
});

export const miscQueue = new Queue<MiscQueuePayload>(getQueueName('misc'), {
  connection: getRedisQueue(),
  defaultJobOptions: {
    removeOnComplete: 10,
  },
});

export type NotificationQueuePayload = {
  type: 'sendNotification';
  payload: {
    notification: Prisma.NotificationUncheckedCreateInput;
  };
};

export const notificationQueue = new Queue<NotificationQueuePayload>(
  getQueueName('notification'),
  {
    connection: getRedisQueue(),
    defaultJobOptions: {
      removeOnComplete: 10,
    },
  }
);

export type ImportQueuePayload = {
  type: 'import';
  payload: {
    importId: string;
  };
};

export const importQueue = new Queue<ImportQueuePayload>(
  getQueueName('import'),
  {
    connection: getRedisQueue(),
    defaultJobOptions: {
      removeOnComplete: 10,
      removeOnFail: 50,
    },
  }
);

export type InsightsQueuePayloadProject = {
  type: 'insightsProject';
  payload: { projectId: string; date: string };
};

export const insightsQueue = new Queue<InsightsQueuePayloadProject>(
  getQueueName('insights'),
  {
    connection: getRedisQueue(),
    defaultJobOptions: {
      removeOnComplete: 100,
    },
  }
);

export type GscQueuePayloadSync = {
  type: 'gscProjectSync';
  payload: { projectId: string };
};
export type GscQueuePayloadBackfill = {
  type: 'gscProjectBackfill';
  payload: { projectId: string };
};
export type GscQueuePayload = GscQueuePayloadSync | GscQueuePayloadBackfill;

export const gscQueue = new Queue<GscQueuePayload>(getQueueName('gsc'), {
  connection: getRedisQueue(),
  defaultJobOptions: {
    removeOnComplete: 50,
    removeOnFail: 100,
  },
});