Method ingest

packages/db/src/buffers/session-buffer.ts:132–169

* Ingest one event into the session lifecycle. * * Reads the device's current session (if any), decides whether the event * extends it, opens a brand-new one, or boundaries (gap > 30min) → close * the old session and start a fresh one. Writes the updated session blob, * updates the wa

(
    payload: IServiceCreateEventPayload
  )

   * themselves are skipped — they are derived signals, not session activity.
   */
  async ingest(
    payload: IServiceCreateEventPayload
  ): Promise<SessionIngestResult | null> {
    if (!payload.projectId || !payload.deviceId) return null;
    if (payload.name === 'session_start' || payload.name === 'session_end') {
      return null;
    }

    try {
      const existing = await this.getExistingSession({
        projectId: payload.projectId,
        deviceId: payload.deviceId,
      });

      const eventTimeMs = payload.createdAt.getTime();
      const isBoundary =
        existing &&
        eventTimeMs - fromClickhouseDate(existing.ended_at).getTime() >
          SESSION_TIMEOUT_MS;

      if (existing && !isBoundary) {
        const { current, chRows } = this.extendSession(existing, payload);
        await this.persist(current, chRows);
        return { kind: 'extend', current };
      }

      const current = this.newSession(payload);
      await this.persist(current, [current]);

      if (existing && isBoundary) {
        return { kind: 'boundary', current, closed: existing };
      }
      return { kind: 'new', current };
    } catch (error) {
      this.logger.error({ err: error }, 'Failed to ingest session');
      return null;
    }
  }

  /**
   * Remove a closed session from Redis. Atomically gated on `sessionId` via