stats/apps/api/src/controllers/track.controller.ts

import type { FastifyReply, FastifyRequest } from 'fastify';
import { assocPath, pathOr, pick } from 'ramda';

import { HttpError } from '@/utils/errors';
import { generateId, slug } from '@openpanel/common';
import { generateDeviceId, parseUserAgent } from '@openpanel/common/server';
import { getProfileById, getSalts, upsertProfile } from '@openpanel/db';
import { type GeoLocation, getGeoLocation } from '@openpanel/geo';
import { getEventsGroupQueueShard } from '@openpanel/queue';
import { getRedisCache } from '@openpanel/redis';
import type {
  DecrementPayload,
  IdentifyPayload,
  IncrementPayload,
  TrackHandlerPayload,
  TrackPayload,
} from '@openpanel/sdk';

export function getStringHeaders(headers: FastifyRequest['headers']) {
  return Object.entries(
    pick(
      [
        'user-agent',
        'openpanel-sdk-name',
        'openpanel-sdk-version',
        'openpanel-client-id',
        'request-id',
      ],
      headers,
    ),
  ).reduce(
    (acc, [key, value]) => ({
      ...acc,
      [key]: value ? String(value) : undefined,
    }),
    {},
  );
}

function getIdentity(body: TrackHandlerPayload): IdentifyPayload | undefined {
  const identity =
    'properties' in body.payload
      ? (body.payload?.properties?.__identify as IdentifyPayload | undefined)
      : undefined;

  return (
    identity ||
    (body?.payload?.profileId
      ? {
          profileId: body.payload.profileId,
        }
      : undefined)
  );
}

export function getTimestamp(
  timestamp: FastifyRequest['timestamp'],
  payload: TrackHandlerPayload['payload'],
) {
  const safeTimestamp = timestamp || Date.now();
  const userDefinedTimestamp =
    'properties' in payload
      ? (payload?.properties?.__timestamp as string | undefined)
      : undefined;

  if (!userDefinedTimestamp) {
    return { timestamp: safeTimestamp, isTimestampFromThePast: false };
  }

  const clientTimestamp = new Date(userDefinedTimestamp);
  const clientTimestampNumber = clientTimestamp.getTime();

  // Constants for time validation
  const ONE_MINUTE_MS = 60 * 1000;
  const FIFTEEN_MINUTES_MS = 15 * ONE_MINUTE_MS;

  // Use safeTimestamp if invalid or more than 1 minute in the future
  if (
    Number.isNaN(clientTimestampNumber) ||
    clientTimestampNumber > safeTimestamp + ONE_MINUTE_MS
  ) {
    return { timestamp: safeTimestamp, isTimestampFromThePast: false };
  }

  // isTimestampFromThePast is true only if timestamp is older than 1 hour
  const isTimestampFromThePast =
    clientTimestampNumber < safeTimestamp - FIFTEEN_MINUTES_MS;

  return {
    timestamp: clientTimestampNumber,
    isTimestampFromThePast,
  };
}

export async function handler(
  request: FastifyRequest<{
    Body: TrackHandlerPayload;
  }>,
  reply: FastifyReply,
) {
  const timestamp = getTimestamp(request.timestamp, request.body.payload);
  const ip =
    'properties' in request.body.payload &&
    request.body.payload.properties?.__ip
      ? (request.body.payload.properties.__ip as string)
      : request.clientIp;
  const ua = request.headers['user-agent'];
  const projectId = request.client?.projectId;

  if (!projectId) {
    return reply.status(400).send({
      status: 400,
      error: 'Bad Request',
      message: 'Missing projectId',
    });
  }

  const identity = getIdentity(request.body);
  const profileId = identity?.profileId;
  const overrideDeviceId = (() => {
    const deviceId =
      'properties' in request.body.payload
        ? request.body.payload.properties?.__deviceId
        : undefined;
    if (typeof deviceId === 'string') {
      return deviceId;
    }
    return undefined;
  })();

  // We might get a profileId from the alias table
  // If we do, we should use that instead of the one from the payload
  if (profileId) {
    request.body.payload.profileId = profileId;
  }

  switch (request.body.type) {
    case 'track': {
      const [salts, geo] = await Promise.all([getSalts(), getGeoLocation(ip)]);
      const currentDeviceId =
        overrideDeviceId ||
        (ua
          ? generateDeviceId({
              salt: salts.current,
              origin: projectId,
              ip,
              ua,
            })
          : '');
      const previousDeviceId = ua
        ? generateDeviceId({
            salt: salts.previous,
            origin: projectId,
            ip,
            ua,
          })
        : '';

      const promises = [];

      // If we have more than one property in the identity object, we should identify the user
      // Otherwise its only a profileId and we should not identify the user
      if (identity && Object.keys(identity).length > 1) {
        promises.push(
          identify({
            payload: identity,
            projectId,
            geo,
            ua,
          }),
        );
      }

      promises.push(
        track({
          payload: request.body.payload,
          currentDeviceId,
          previousDeviceId,
          projectId,
          geo,
          headers: getStringHeaders(request.headers),
          timestamp: timestamp.timestamp,
          isTimestampFromThePast: timestamp.isTimestampFromThePast,
        }),
      );

      await Promise.all(promises);
      break;
    }
    case 'identify': {
      const payload = request.body.payload;
      const geo = await getGeoLocation(ip);
      if (!payload.profileId) {
        throw new HttpError('Missing profileId', {
          status: 400,
        });
      }

      await identify({
        payload,
        projectId,
        geo,
        ua,
      });
      break;
    }
    case 'alias': {
      return reply.status(400).send({
        status: 400,
        error: 'Bad Request',
        message: 'Alias is not supported',
      });
    }
    case 'increment': {
      await increment({
        payload: request.body.payload,
        projectId,
      });
      break;
    }
    case 'decrement': {
      await decrement({
        payload: request.body.payload,
        projectId,
      });
      break;
    }
    default: {
      return reply.status(400).send({
        status: 400,
        error: 'Bad Request',
        message: 'Invalid type',
      });
    }
  }

  reply.status(200).send();
}

async function track({
  payload,
  currentDeviceId,
  previousDeviceId,
  projectId,
  geo,
  headers,
  timestamp,
  isTimestampFromThePast,
}: {
  payload: TrackPayload;
  currentDeviceId: string;
  previousDeviceId: string;
  projectId: string;
  geo: GeoLocation;
  headers: Record<string, string | undefined>;
  timestamp: number;
  isTimestampFromThePast: boolean;
}) {
  const uaInfo = parseUserAgent(headers['user-agent'], payload.properties);
  const groupId = uaInfo.isServer
    ? payload.profileId
      ? `${projectId}:${payload.profileId}`
      : `${projectId}:${generateId()}`
    : currentDeviceId;
  const jobId = [
    slug(payload.name),
    timestamp,
    projectId,
    currentDeviceId,
    groupId,
  ]
    .filter(Boolean)
    .join('-');
  await getEventsGroupQueueShard(groupId).add({
    orderMs: timestamp,
    data: {
      projectId,
      headers,
      event: {
        ...payload,
        timestamp,
        isTimestampFromThePast,
      },
      uaInfo,
      geo,
      currentDeviceId,
      previousDeviceId,
    },
    groupId,
    jobId,
  });
}

async function identify({
  payload,
  projectId,
  geo,
  ua,
}: {
  payload: IdentifyPayload;
  projectId: string;
  geo: GeoLocation;
  ua?: string;
}) {
  const uaInfo = parseUserAgent(ua, payload.properties);
  await upsertProfile({
    ...payload,
    id: payload.profileId,
    isExternal: true,
    projectId,
    properties: {
      ...(payload.properties ?? {}),
      country: geo.country,
      city: geo.city,
      region: geo.region,
      longitude: geo.longitude,
      latitude: geo.latitude,
      os: uaInfo.os,
      os_version: uaInfo.osVersion,
      browser: uaInfo.browser,
      browser_version: uaInfo.browserVersion,
      device: uaInfo.device,
      brand: uaInfo.brand,
      model: uaInfo.model,
    },
  });
}

async function increment({
  payload,
  projectId,
}: {
  payload: IncrementPayload;
  projectId: string;
}) {
  const { profileId, property, value } = payload;
  const profile = await getProfileById(profileId, projectId);
  if (!profile) {
    throw new Error('Not found');
  }

  const parsed = Number.parseInt(
    pathOr<string>('0', property.split('.'), profile.properties),
    10,
  );

  if (Number.isNaN(parsed)) {
    throw new Error('Not number');
  }

  profile.properties = assocPath(
    property.split('.'),
    parsed + (value || 1),
    profile.properties,
  );

  await upsertProfile({
    id: profile.id,
    projectId,
    properties: profile.properties,
    isExternal: true,
  });
}

async function decrement({
  payload,
  projectId,
}: {
  payload: DecrementPayload;
  projectId: string;
}) {
  const { profileId, property, value } = payload;
  const profile = await getProfileById(profileId, projectId);
  if (!profile) {
    throw new Error('Not found');
  }

  const parsed = Number.parseInt(
    pathOr<string>('0', property.split('.'), profile.properties),
    10,
  );

  if (Number.isNaN(parsed)) {
    throw new Error('Not number');
  }

  profile.properties = assocPath(
    property.split('.'),
    parsed - (value || 1),
    profile.properties,
  );

  await upsertProfile({
    id: profile.id,
    projectId,
    properties: profile.properties,
    isExternal: true,
  });
}

export async function fetchDeviceId(
  request: FastifyRequest,
  reply: FastifyReply,
) {
  const salts = await getSalts();
  const projectId = request.client?.projectId;
  if (!projectId) {
    return reply.status(400).send('No projectId');
  }

  const ip = request.clientIp;
  if (!ip) {
    return reply.status(400).send('Missing ip address');
  }

  const ua = request.headers['user-agent'];
  if (!ua) {
    return reply.status(400).send('Missing header: user-agent');
  }

  const currentDeviceId = generateDeviceId({
    salt: salts.current,
    origin: projectId,
    ip,
    ua,
  });
  const previousDeviceId = generateDeviceId({
    salt: salts.previous,
    origin: projectId,
    ip,
    ua,
  });

  try {
    const multi = getRedisCache().multi();
    multi.exists(`bull:sessions:sessionEnd:${projectId}:${currentDeviceId}`);
    multi.exists(`bull:sessions:sessionEnd:${projectId}:${previousDeviceId}`);
    const res = await multi.exec();

    if (res?.[0]?.[1]) {
      return reply.status(200).send({
        deviceId: currentDeviceId,
        message: 'current session exists for this device id',
      });
    }

    if (res?.[1]?.[1]) {
      return reply.status(200).send({
        deviceId: previousDeviceId,
        message: 'previous session exists for this device id',
      });
    }
  } catch (error) {
    request.log.error('Error getting session end GET /track/device-id', error);
  }

  return reply.status(200).send({
    deviceId: currentDeviceId,
    message: 'No session exists for this device id',
  });
}