import type lambda from "aws-lambda";
import * as Effect from "effect/Effect";
import * as Layer from "effect/Layer";
import * as Stream from "effect/Stream";

import * as Namespace from "../../Namespace.ts";
import type { Bucket } from "../S3/Bucket.ts";
import {
  BucketEventSource as S3BucketEventSource,
  type BucketEventSourceService,
} from "../S3/BucketEventSource.ts";
import type {
  BucketNotification,
  NotificationsProps,
} from "../S3/BucketNotifications.ts";
import type { S3EventType } from "../S3/S3Event.ts";
import * as Lambda from "./Function.ts";
import { Permission as LambdaPermission } from "./Permission.ts";

/**
 * Connects an S3 bucket notification stream to the current Lambda function.
 *
 * This layer listens for bucket notifications routed through the Lambda runtime
 * and exposes them as an `Effect.Stream`, while the companion policy configures
 * the invoke permission and bucket notification binding during deployment.
 * ### Wiring Events
 * **Example:** Listen for Object Created Events
 * ```typescript
 * yield* AWS.Lambda.BucketEventSource(
 *   bucket,
 *   { events: ["s3:ObjectCreated:*"] },
 *   (events) => Stream.runForEach(events, (event) => Effect.log(event.key)),
 * );
 * ```
 *
 * @binding
 */
export const BucketEventSource = Layer.effect(
  S3BucketEventSource,
  Effect.gen(function* () {
    // this layer can only be used in a Lambda Function
    const func = yield* Lambda.Function;
    const Permission = yield* LambdaPermission;

    return Effect.fn(function* <
      Events extends S3EventType[],
      StreamReq = never,
      Req = never,
    >(
      bucket: Bucket,
      props: NotificationsProps<Events>,
      process: (
        stream: Stream.Stream<BucketNotification, never, StreamReq>,
      ) => Effect.Effect<void, never, Req>,
    ) {
      // this adds it to the Lambda Function's environment variables
      const BucketName = yield* bucket.bucketName;

      // Deploy-time: grant the invoke permission and attach the bucket
      // notification config. Skipped once running inside the deployed Function
      // (the global guard), where the only work is registering the runtime
      // handler below. Namespaced under the host so the sub-resources' logical
      // identity matches the previous Binding.Policy.
      if (!globalThis.__ALCHEMY_RUNTIME__) {
        yield* Namespace.push(
          func.LogicalId,
          Effect.gen(function* () {
            const {
              events: Events = ["s3:ObjectCreated:*"],
              prefix,
              suffix,
            } = props ?? {};
            const filterRules = [
              ...(prefix !== undefined
                ? [{ Name: "prefix" as const, Value: prefix }]
                : []),
              ...(suffix !== undefined
                ? [{ Name: "suffix" as const, Value: suffix }]
                : []),
            ];
            yield* Permission(
              `AWS.Lambda.InvokeFunction(${bucket.LogicalId})`,
              {
                action: "lambda:InvokeFunction",
                functionName: func.functionName,
                principal: "s3.amazonaws.com",
                sourceArn: bucket.bucketArn,
              },
            );
            yield* bucket.bind(`AWS.S3.Notifications(${bucket.LogicalId})`, {
              notificationConfiguration: {
                LambdaFunctionConfigurations: [
                  {
                    LambdaFunctionArn: func.functionArn,
                    Events,
                    ...(filterRules.length > 0
                      ? { Filter: { Key: { FilterRules: filterRules } } }
                      : {}),
                  },
                ],
              },
            });
          }),
        );
      }

      yield* func.listen(
        Effect.gen(function* () {
          // this accesses it
          const bucketName = yield* BucketName;
          return (event: any) => {
            if (isS3Event(event)) {
              const events = event.Records.filter(
                (record) => record.s3.bucket.name === bucketName,
              );
              if (events.length > 0) {
                return process(
                  Stream.fromArray(
                    events.map((record: lambda.S3EventRecord) => ({
                      type: record.eventName as S3EventType,
                      bucket: record.s3.bucket.name,
                      key: record.s3.object.key,
                      size: record.s3.object.size,
                      eTag: record.s3.object.eTag,
                    })),
                  ),
                  // TODO(sam): don't die?
                ).pipe(Effect.orDie);
              }
            }
          };
        }),
      );
    }) as BucketEventSourceService;
  }),
);

const isS3Event = (event: any): event is lambda.S3Event =>
  Array.isArray(event.Records) &&
  event.Records.some((record: any) => record.s3);