feat: improve error handling for authentication flow

src/controllers/auth/anilist/getWatchingTitles.ts

@@ -1,6 +1,8 @@
 import { graphql } from "gql.tada";
 import { GraphQLClient } from "graphql-request";
 
+import { sleep } from "~/libs/sleep";
+
 const GetWatchingTitlesQuery = graphql(`
   query GetWatchingTitles($userName: String!, $page: Int!) {
     Page(page: $page, perPage: 50) {
@@ -55,8 +57,7 @@ export function getWatchingTitles(
   username: string,
   page: number,
   aniListToken: string,
-  executionCtx: ExecutionContext,
+): Promise<GetWatchingTitles> {
-) {
   const client = new GraphQLClient("https://graphql.anilist.co/");
 
   return client
@@ -73,16 +74,67 @@ export function getWatchingTitles(
       const response = err.response;
       if (response.status === 429) {
         console.log("429, retrying in", response.headers.get("Retry-After"));
-        executionCtx.waitUntil(
+        return sleep(Number(response.headers.get("Retry-After")!) * 1000).then(
-          sleep(Number(response.headers.get("Retry-After")!) * 1000),
+          () => getWatchingTitles(username, page, aniListToken),
         );
-        return getWatchingTitles(username, page, aniListToken, executionCtx);
       }
 
       throw err;
     });
 }
 
-function sleep(ms: number) {
+type GetWatchingTitles = {
-  return new Promise((resolve) => setTimeout(resolve, ms));
+  mediaList:
-}
+    | ({
+        media: {
+          id: number;
+          idMal: number | null;
+          title: {
+            english: string | null;
+            userPreferred: string | null;
+          } | null;
+          description: string | null;
+          episodes: number | null;
+          genres: (string | null)[] | null;
+          status:
+            | "FINISHED"
+            | "RELEASING"
+            | "NOT_YET_RELEASED"
+            | "CANCELLED"
+            | "HIATUS"
+            | null;
+          bannerImage: string | null;
+          averageScore: number | null;
+          coverImage: {
+            extraLarge: string | null;
+            large: string | null;
+            medium: string | null;
+          } | null;
+          countryOfOrigin: unknown;
+          mediaListEntry: {
+            id: number;
+            progress: number | null;
+            status:
+              | "CURRENT"
+              | "REPEATING"
+              | "PLANNING"
+              | "COMPLETED"
+              | "DROPPED"
+              | "PAUSED"
+              | null;
+          } | null;
+          nextAiringEpisode: {
+            timeUntilAiring: number;
+            airingAt: number;
+            episode: number;
+          } | null;
+        } | null;
+      } | null)[]
+    | null;
+  pageInfo: {
+    currentPage: number | null;
+    hasNextPage: boolean | null;
+    perPage: number | null;
+    total: number | null;
+  } | null;
+};

src/controllers/auth/anilist/index.ts

@@ -4,11 +4,11 @@ import { streamSSE } from "hono/streaming";
 
 import { fetchEpisodes } from "~/controllers/episodes/getByAniListId";
 import { maybeScheduleNextAiringEpisode } from "~/libs/maybeScheduleNextAiringEpisode";
-import { readEnvVariable } from "~/libs/readEnvVariable";
+import { sleep } from "~/libs/sleep";
 import { associateDeviceIdWithUsername } from "~/models/token";
 import { setWatchStatus } from "~/models/watchStatus";
 import type { Env } from "~/types/env";
-import { EpisodesResponseSchema } from "~/types/episode";
+import { Episode, EpisodesResponseSchema } from "~/types/episode";
 import { ErrorResponse, ErrorResponseSchema } from "~/types/schema";
 import { Title } from "~/types/title";
 
@@ -137,7 +137,6 @@ app.openapi(route, async (c) => {
           user.name!,
           currentPage++,
           aniListToken,
-          c.executionCtx,
         );
         if (!mediaList) {
           break;
@@ -151,7 +150,7 @@ app.openapi(route, async (c) => {
         }
 
         for (const mediaObj of mediaList) {
-          const media = mediaObj?.media!;
+          const media = mediaObj?.media;
           if (!media) {
             continue;
           }
@@ -189,8 +188,14 @@ app.openapi(route, async (c) => {
 
           await fetchEpisodes(
             media.id,
-            readEnvVariable<boolean>(c.env, "ENABLE_ANIFY"),
+            { ...env(c, "workerd"), ENABLE_ANIFY: "false" },
-          ).then(({ result: { episodes } }) => {
+            true,
+          ).then(({ result: episodesResult }) => {
+            const episodes = episodesResult?.episodes;
+            if (!episodes) {
+              return;
+            }
+
             stream.writeSSE({
               event: "title",
               data: JSON.stringify({ title: media, episodes }),
@@ -199,15 +204,13 @@ app.openapi(route, async (c) => {
           });
         }
 
-        hasNextPage = pageInfo?.hasNextPage ?? false;
-        hasNextPage = pageInfo?.hasNextPage ?? false;
-        console.log(hasNextPage);
         hasNextPage = pageInfo?.hasNextPage ?? false;
         console.log(hasNextPage);
       } while (hasNextPage);
 
       // send end event instead of closing the connection to let the client know that the stream didn't end abruptly
       await stream.writeSSE({ event: "end", data: "end" });
+      console.log("completed");
     },
     async (err, stream) => {
       console.error("Error occurred in SSE");

src/controllers/episodes/getByAniListId/anify.ts

@@ -1,15 +1,18 @@
-import { z } from "zod";
+import { DateTime } from "luxon";
 
 import { PromiseTimedOutError, promiseTimeout } from "~/libs/promiseTimeout";
+import { readEnvVariable } from "~/libs/readEnvVariable";
 import { sortByProperty } from "~/libs/sortByProperty";
-
+import { getValue, setValue } from "~/models/kv";
-import type { EpisodesResponse } from "./episode";
+import type { Env } from "~/types/env";
+import type { EpisodesResponse } from "~/types/episode";
 
 export async function getEpisodesFromAnify(
-  isAnifyEnabled: boolean,
+  env: Env,
   aniListId: number,
 ): Promise<EpisodesResponse | null> {
-  if (shouldSkipAnify(isAnifyEnabled, aniListId)) {
+  if (await shouldSkipAnify(env, aniListId)) {
+    console.log("Skipping Anify for title", aniListId);
     return null;
   }
 
@@ -19,9 +22,25 @@ export async function getEpisodesFromAnify(
     response = await promiseTimeout(
       fetch(`https://anify.eltik.cc/episodes/${aniListId}`, {
         signal: abortController.signal,
-      }).then((res) => res.json<AnifyEpisodesResponse[]>()),
+      }).then((res) => res.json() as Promise<AnifyEpisodesResponse[]>),
       30 * 1000,
     );
+    if ("error" in response) {
+      const error = response.error;
+      if (error === "Too many requests") {
+        console.log(
+          "Sending too many requests to Anify, setting killswitch until",
+          DateTime.now().plus({ minutes: 1 }).toISO(),
+        );
+        setValue(
+          env,
+          "anify_killswitch_till",
+          DateTime.now().plus({ minutes: 1 }).toISO(),
+        );
+      }
+
+      return null;
+    }
   } catch (e) {
     if (e instanceof PromiseTimedOutError) {
       abortController.abort("Loading episodes from Anify timed out");
@@ -73,11 +92,11 @@ export async function getEpisodesFromAnify(
   };
 }
 
-export function shouldSkipAnify(
+export async function shouldSkipAnify(
-  isAnifyEnabled: boolean,
+  env: Env,
   aniListId: number,
-): boolean {
+): Promise<boolean> {
-  if (!isAnifyEnabled) {
+  if (!readEnvVariable(env, "ENABLE_ANIFY")) {
     return true;
   }
 
@@ -92,7 +111,13 @@ export function shouldSkipAnify(
     return true;
   }
 
-  return false;
+  return await getValue(env, "anify_killswitch_till").then((dateTime) => {
+    if (!dateTime) {
+      return false;
+    }
+
+    return DateTime.fromISO(dateTime).diffNow().as("minutes") > 0;
+  });
 }
 
 interface AnifyEpisodesResponse {

src/controllers/episodes/getByAniListId/aniwatch.ts

@@ -1,8 +1,10 @@
 import { findBestMatchingTitle } from "~/libs/findBestMatchingTitle";
+import { sleep } from "~/libs/sleep";
 import { Episode, type EpisodesResponse } from "~/types/episode";
 
 export async function getEpisodesFromAniwatch(
   aniListId: number,
+  shouldRetry: boolean = false,
 ): Promise<EpisodesResponse | null> {
   try {
     const animeTitle = await import("~/libs/anilist/getTitle")
@@ -48,6 +50,16 @@ export async function getEpisodesFromAniwatch(
 
     return { providerId: "aniwatch", episodes };
   } catch (error) {
+    if (shouldRetry && "response" in error && error.response.status === 429) {
+      console.log(
+        "429, retrying in",
+        error.response.headers.get("Retry-After"),
+      );
+      return sleep(
+        Number(error.response.headers.get("Retry-After")!) * 1000,
+      ).then(() => getEpisodesFromAniwatch(aniListId));
+    }
+
     console.error(
       new Error(
         `Error trying to load episodes from aniwatch; aniListId: ${aniListId}`,

src/controllers/episodes/getByAniListId/index.ts

@@ -1,4 +1,5 @@
 import { OpenAPIHono, createRoute, z } from "@hono/zod-openapi";
+import { env } from "hono/adapter";
 
 import { fetchFromMultipleSources } from "~/libs/fetchFromMultipleSources";
 import { readEnvVariable } from "~/libs/readEnvVariable";
@@ -45,13 +46,13 @@ const app = new OpenAPIHono<Env>();
 
 export function fetchEpisodesFromAllProviders(
   aniListId: number,
-  isAnifyEnabled: boolean,
+  env: Env,
 ): Promise<EpisodesResponse[]> {
   return Promise.allSettled([
     import("./aniwatch").then(({ getEpisodesFromAniwatch }) =>
       getEpisodesFromAniwatch(aniListId),
     ),
-    getEpisodesFromAnify(isAnifyEnabled, aniListId),
+    getEpisodesFromAnify(env, aniListId),
   ]).then((episodeResults) =>
     episodeResults
       .filter((result) => result.status === "fulfilled")
@@ -60,16 +61,20 @@ export function fetchEpisodesFromAllProviders(
   );
 }
 
-export function fetchEpisodes(aniListId: number, isAnifyEnabled: boolean) {
+export function fetchEpisodes(
+  aniListId: number,
+  env: Env,
+  shouldRetry: boolean = false,
+) {
   return fetchFromMultipleSources([
-    () => getEpisodesFromAnify(isAnifyEnabled, aniListId),
+    () => getEpisodesFromAnify(env, aniListId),
     // () =>
     //   import("./consumet").then(({ getEpisodesFromConsumet }) =>
     //     getEpisodesFromConsumet(aniListId),
     //   ),
     () =>
       import("./aniwatch").then(({ getEpisodesFromAniwatch }) =>
-        getEpisodesFromAniwatch(aniListId),
+        getEpisodesFromAniwatch(aniListId, shouldRetry),
       ),
   ]);
 }
@@ -79,7 +84,7 @@ app.openapi(route, async (c) => {
 
   const { result: episodes, errorOccurred } = await fetchEpisodes(
     aniListId,
-    readEnvVariable<boolean>(c.env, "ENABLE_ANIFY"),
+    env(c, "workerd"),
   );
 
   if (errorOccurred) {

src/controllers/episodes/getEpisodeUrl/index.ts

@@ -1,4 +1,5 @@
 import { OpenAPIHono, createRoute, z } from "@hono/zod-openapi";
+import { env } from "hono/adapter";
 
 import { readEnvVariable } from "~/libs/readEnvVariable";
 import type { Env } from "~/types/env";
@@ -75,12 +76,9 @@ const app = new OpenAPIHono<Env>();
 export async function fetchEpisodeUrlFromAllProviders(
   aniListId: number,
   episodeNumber: number,
-  isAnifyEnabled: boolean,
+  env: Env,
 ) {
-  const results = await fetchEpisodesFromAllProviders(
+  const results = await fetchEpisodesFromAllProviders(aniListId, env);
-    aniListId,
-    isAnifyEnabled,
-  );
   if (results.length === 0) {
     return { episodes: null, fetchUrlResult: null };
   }
@@ -101,7 +99,7 @@ export async function fetchEpisodeUrlFromAllProviders(
       providerId,
       episode.id,
       aniListId,
-      isAnifyEnabled,
+      readEnvVariable(env, "ENABLE_ANIFY"),
     );
     if (!urlResult) {
       episodes = null;
@@ -190,7 +188,7 @@ app.openapi(route, async (c) => {
       const { fetchUrlResult } = await fetchEpisodeUrlFromAllProviders(
         aniListId,
         episodeNumber!,
-        isAnifyEnabled,
+        env(c, "workerd"),
       );
       if (!fetchUrlResult) {
         return c.json(ErrorResponse, { status: 404 });

src/controllers/internal/new-episode/index.ts

@@ -53,14 +53,10 @@ app.post(
       );
     }
 
-    const isAnifyEnabled = readEnvVariable<boolean>(
-      env<Env, typeof c>(c, "workerd"),
-      "ENABLE_ANIFY",
-    );
     const { episodes, fetchUrlResult } = await fetchEpisodeUrlFromAllProviders(
       aniListId,
       episodeNumber,
-      isAnifyEnabled,
+      env<Env, typeof c>(c, "workerd"),
     );
 
     if (!episodes) {

src/libs/sleep.ts

@@ -0,0 +1,3 @@
+export function sleep(ms: number) {
+  return new Promise((resolve) => setTimeout(resolve, ms));
+}

src/models/schema.ts

@@ -30,7 +30,9 @@ export const watchStatusTable = sqliteTable(
 );
 
 export const keyValueTable = sqliteTable("key_value", {
-  key: text("key", { enum: ["schedule_last_checked_at"] }).primaryKey(),
+  key: text("key", {
+    enum: ["schedule_last_checked_at", "anify_killswitch_till"],
+  }).primaryKey(),
   value: text("value").notNull(),
 });