Function layer

packages/core/analytics/src/analytics.ts:151–236

(
  config: AnalyticsConfig,
)

}

export const layer = (
  config: AnalyticsConfig,
): Layer.Layer<Analytics, never, HttpClient.HttpClient | FileSystem.FileSystem> =>
  Layer.effect(Analytics)(
    Effect.gen(function* () {
      const disabled = config.disabled ?? isAnalyticsDisabled();
      if (disabled) return analyticsNoop;

      const http = yield* HttpClient.HttpClient;
      const distinctId = yield* loadOrMintAnonymousId(config.dataDir);
      const buffer = yield* Ref.make<ReadonlyArray<BufferedEvent>>([]);

      const posthogHost = config.posthogHost ?? DEFAULT_POSTHOG_HOST;
      const posthogKey = config.posthogKey ?? DEFAULT_POSTHOG_KEY;

      const sendBatch = (events: ReadonlyArray<BufferedEvent>) =>
        HttpClientRequest.post(`${posthogHost}/batch/`).pipe(
          HttpClientRequest.bodyJson({
            api_key: posthogKey,
            batch: events.map((entry) => ({
              event: entry.event,
              distinct_id: distinctId,
              timestamp: entry.timestamp,
              properties: {
                ...entry.properties,
                // Anonymous events only: never materialize a person profile.
                $process_person_profile: false,
                surface: config.surface,
                channel: config.channel,
                app_version: config.version,
              },
            })),
          }),
          Effect.flatMap(http.execute),
          Effect.flatMap(HttpClientResponse.filterStatusOk),
          Effect.asVoid,
          Effect.timeout(Duration.seconds(10)),
        );

      const flush: Effect.Effect<void> = Effect.gen(function* () {
        while (true) {
          const batch = yield* Ref.modify(buffer, (current) => {
            const next = current.slice(0, FLUSH_BATCH_SIZE);
            return [next, current.slice(next.length)] as const;
          });
          if (batch.length === 0) return;
          yield* sendBatch(batch).pipe(
            // Delivery failed: put the batch back for the next cadence tick and
            // stop draining (the endpoint is unreachable, later batches would
            // fail the same way).
            Effect.tapError(() => Ref.update(buffer, (current) => [...batch, ...current])),
          );
        }
      }).pipe(
        Effect.catchCause((cause) =>
          Effect.logDebug("Analytics.flush failed").pipe(Effect.annotateLogs("cause", cause)),
        ),
      );