import {
  createTrigger,
  TriggerStrategy,
  FilesService,
} from '@activepieces/pieces-framework';
import { GmailLabel } from '../common/models';
import { GmailProps } from '../common/props';
import { gmailAuth, createGoogleClient, GmailAuthValue } from '../auth';
import {
  parseStream,
  convertAttachment,
  getFirstFiveOrAll,
} from '../common/data';
import { gmail as googleGmail } from '@googleapis/gmail';

export const gmailNewEmailTrigger = createTrigger({
  auth: gmailAuth,
  name: 'gmail_new_email_received',
  displayName: 'New Email',
  description: 'Triggers when new mail is found in your Gmail inbox',
  aiMetadata: {
    description:
      'Fires when a new email arrives in the connected Gmail account, optionally narrowed by sender, recipient, subject, label, or category filters. Each event represents one newly received message with its parsed contents and thread.',
  },
  props: {
    subject: GmailProps.subject,
    from: GmailProps.from,
    to: GmailProps.to,
    label: GmailProps.label({ required: false }),
    category: GmailProps.category,
  },
  sampleData: {},
  type: TriggerStrategy.POLLING,
  async onEnable(context) {
    await context.store.put('lastPoll', Date.now());
  },
  async onDisable(context) {
    return;
  },
  async run(context) {
    const lastFetchEpochMS = (await context.store.get<number>('lastPoll')) ?? 0;

    const items = await pollRecentMessages({
      auth: context.auth,
      props: context.propsValue,
      files: context.files,
      lastFetchEpochMS: lastFetchEpochMS,
    });

    const newLastEpochMilliSeconds = items.reduce(
      (acc, item) => Math.max(acc, item.epochMilliSeconds),
      lastFetchEpochMS
    );
    await context.store.put('lastPoll', newLastEpochMilliSeconds);
    return items
      .filter((f) => f.epochMilliSeconds > lastFetchEpochMS)
      .map((item) => item.data);
  },
  async test(context) {
    const lastFetchEpochMS = (await context.store.get<number>('lastPoll')) ?? 0;

    const items = await pollRecentMessages({
      auth: context.auth,
      props: context.propsValue,
      files: context.files,
      lastFetchEpochMS: lastFetchEpochMS,
    });

    return getFirstFiveOrAll(items.map((item) => item.data));
  },
});

interface PropsValue {
  from: string | undefined;
  to: string | undefined;
  subject: string | undefined;
  label: GmailLabel | undefined;
  category: string | undefined;
}

async function pollRecentMessages({
  auth,
  props,
  files,
  lastFetchEpochMS,
}: {
  auth: GmailAuthValue;
  props: PropsValue;
  files: FilesService;
  lastFetchEpochMS: number;
}): Promise<
  {
    epochMilliSeconds: number;
    data: unknown;
  }[]
> {
  const authClient = await createGoogleClient(auth);

  const gmail = googleGmail({ version: 'v1', auth: authClient });

  // construct query
  const query = [];
  const maxResults = lastFetchEpochMS === 0 ? 5 : 20;
  const afterUnixSeconds = Math.floor(lastFetchEpochMS / 1000);

  if (props.from) query.push(`from:(${props.from})`);
  if (props.to) query.push(`to:(${props.to})`);
  if (props.subject) query.push(`subject:(${props.subject})`);
  if (props.label) query.push(`label:${props.label.name}`);
  if (props.category) query.push(`category:${props.category}`);
  if (afterUnixSeconds != null && afterUnixSeconds > 0)
    query.push(`after:${afterUnixSeconds}`);

  // List Messages
  const messagesResponse = await gmail.users.messages.list({
    userId: 'me',
    q: query.join(' '),
    maxResults,
  });

  // Reverse to process oldest-first so partial progress doesn't skip messages
  const messages = (messagesResponse.data.messages || []).slice().reverse();

  const pollingResponse = [];
  for (const message of messages) {
    try {
      const rawMailResponse = await gmail.users.messages.get({
        userId: 'me',
        id: message.id!,
        format: 'raw',
      });
      const threadResponse = await gmail.users.threads.get({
        userId: 'me',
        id: message.threadId!,
      });

      const parsedMailResponse = await parseStream(
        Buffer.from(rawMailResponse.data.raw as string, 'base64').toString(
          'utf-8'
        )
      );

      pollingResponse.push({
        epochMilliSeconds: Number(rawMailResponse.data.internalDate),
        data: {
          message: {
            ...parsedMailResponse,
            attachments: await convertAttachment(
              parsedMailResponse.attachments,
              files
            ),
          },
          thread: {
            ...threadResponse,
          },
        },
      });
    } catch (error: any) {
      const isRateLimit =
        error.status === 429 ||
        (error.status === 403 &&
          /quota|rate.?limit/i.test(error.message ?? ''));
      if (isRateLimit) {
        break;
      }
      throw error;
    }
  }

  return pollingResponse;
}
