// stream/[id]/route.ts
let connection: Connection | null = null;
let channel: Channel | null = null;

export async function GET(
  req: NextRequest,
  { params }: { params: { id: string } },
) {
  // gets or refreshes user tokens or throws response error
  const tokens = await getTokens(req);

  if (tokens instanceof NextResponse) return tokens;

  channel = await getChannel();

  const exchange = `stream.${params.id}`;
  const session = nanoid();
  const queueName = `${exchange}.${session}`;

  try {
    await channel.assertExchange(exchange, "fanout", { durable: false });

    const { queue } = await channel.assertQueue(queueName, {
      exclusive: true,
      autoDelete: true,
    });
    await channel.bindQueue(queue, exchange, "");

    const stream = new ReadableStream({
      start(controller) {
        channel?.consume(queue, (message) => {
          if (message) {
            controller.enqueue(message.content);
          }
        });
      },
      async cancel() {
        await channel?.cancel(queue);

        await cleanup();
      },
    });

    const headers = new Headers();

    headers.set("Content-Type", "text/event-stream");
    headers.set("Transfer-Encoding", "chunked");

    const response = new NextResponse(stream, {
      headers,
    });

    if (typeof tokens.data !== "string") {
      response.cookies.set("access_token", tokens.data.at.value, {
        ...DEFAULT_COOKIE_OPTIONS,
        expires: tokens.data.at.expires,
      });

      response.cookies.set("refresh_token", tokens.data.rt.value, {
        ...DEFAULT_COOKIE_OPTIONS,
        expires: tokens.data.rt.expires,
      });
    }

    return response;
  } catch (error) {
    console.error("Error in stream setup", error);

    return NextResponse.json(
      { success: false, error: "Failed to start stream" } as const,
      { status: 500 },
    );
  }
}

// stream/[id]/route.ts
let connection: Connection | null = null;
let channel: Channel | null = null;

export async function GET(
  req: NextRequest,
  { params }: { params: { id: string } },
) {
  // gets or refreshes user tokens or throws response error
  const tokens = await getTokens(req);

  if (tokens instanceof NextResponse) return tokens;

  channel = await getChannel();

  const exchange = `stream.${params.id}`;
  const session = nanoid();
  const queueName = `${exchange}.${session}`;

  try {
    await channel.assertExchange(exchange, "fanout", { durable: false });

    const { queue } = await channel.assertQueue(queueName, {
      exclusive: true,
      autoDelete: true,
    });
    await channel.bindQueue(queue, exchange, "");

    const stream = new ReadableStream({
      start(controller) {
        channel?.consume(queue, (message) => {
          if (message) {
            controller.enqueue(message.content);
          }
        });
      },
      async cancel() {
        await channel?.cancel(queue);

        await cleanup();
      },
    });

    const headers = new Headers();

    headers.set("Content-Type", "text/event-stream");
    headers.set("Transfer-Encoding", "chunked");

    const response = new NextResponse(stream, {
      headers,
    });

    if (typeof tokens.data !== "string") {
      response.cookies.set("access_token", tokens.data.at.value, {
        ...DEFAULT_COOKIE_OPTIONS,
        expires: tokens.data.at.expires,
      });

      response.cookies.set("refresh_token", tokens.data.rt.value, {
        ...DEFAULT_COOKIE_OPTIONS,
        expires: tokens.data.rt.expires,
      });
    }

    return response;
  } catch (error) {
    console.error("Error in stream setup", error);

    return NextResponse.json(
      { success: false, error: "Failed to start stream" } as const,
      { status: 500 },
    );
  }
}