feat: support sending "new episode" notifications to devices

package.json

@@ -20,6 +20,7 @@
     "@haverstack/axios-fetch-adapter": "^0.12.0",
     "@hono/swagger-ui": "^0.2.2",
     "@hono/zod-openapi": "^0.12.0",
+    "@hono/zod-validator": "^0.2.2",
     "@libsql/client": "^0.6.2",
     "@upstash/qstash": "^2.7.0",
     "drizzle-orm": "^0.31.2",

src/controllers/episodes/getByAniListId/index.ts

@@ -3,17 +3,14 @@ import { OpenAPIHono, createRoute, z } from "@hono/zod-openapi";
 import { fetchFromMultipleSources } from "~/libs/fetchFromMultipleSources";
 import { readEnvVariable } from "~/libs/readEnvVariable";
 import type { Env } from "~/types/env";
+import { EpisodesResponseSchema } from "~/types/episode";
 import {
   AniListIdQuerySchema,
   ErrorResponse,
   ErrorResponseSchema,
-  SuccessResponseSchema,
 } from "~/types/schema";
 
 import { getEpisodesFromAnify } from "./anify";
-import { EpisodesResponse } from "./episode";
-
-const EpisodesResponseSchema = SuccessResponseSchema(EpisodesResponse);
 
 const route = createRoute({
   tags: ["aniplay", "episodes"],

src/controllers/episodes/getEpisodeUrl/index.ts

@@ -2,19 +2,18 @@ import { OpenAPIHono, createRoute, z } from "@hono/zod-openapi";
 
 import { readEnvVariable } from "~/libs/readEnvVariable";
 import type { Env } from "~/types/env";
+import {
+  FetchUrlResponse,
+  FetchUrlResponseSchema,
+} from "~/types/episode/fetch-url-response";
 import {
   AniListIdQuerySchema,
   ErrorResponse,
   ErrorResponseSchema,
-  SuccessResponseSchema,
 } from "~/types/schema";
 
-import { FetchUrlResponse as FetchUrlResponseSchema } from "./responseType";
-
 const FetchUrlRequest = z.object({ id: z.string(), provider: z.string() });
 
-const FetchUrlResponse = SuccessResponseSchema(FetchUrlResponseSchema);
-
 const route = createRoute({
   tags: ["aniplay", "episodes"],
   summary: "Fetch stream URL for an episode",

src/controllers/episodes/getEpisodeUrl/responseType.ts

@@ -1,13 +0,0 @@
-import { z } from "zod";
-
-import { SkippableSchema } from "~/types/schema";
-
-export type FetchUrlResponse = z.infer<typeof FetchUrlResponse>;
-export const FetchUrlResponse = z.object({
-  source: z.string(),
-  subtitles: z.array(z.object({ url: z.string(), lang: z.string() })),
-  audio: z.array(z.object({ url: z.string(), lang: z.string() })),
-  intro: SkippableSchema,
-  outro: SkippableSchema,
-  headers: z.record(z.string()).optional(),
-});

src/controllers/new-episode/index.ts

@@ -0,0 +1,111 @@
+import { zValidator } from "@hono/zod-validator";
+import { Hono } from "hono";
+import { env } from "hono/adapter";
+import mapKeys from "lodash.mapkeys";
+import { DateTime } from "luxon";
+import { z } from "zod";
+
+import { Case, changeStringCase } from "~/libs/changeStringCase";
+import type { AdminSdkCredentials } from "~/libs/fcm/getGoogleAuthToken";
+import { sendFcmMessage } from "~/libs/fcm/sendFcmMessage";
+import { verifyQstashHeader } from "~/libs/qstash/verifyQstashHeader";
+import { readEnvVariable } from "~/libs/readEnvVariable";
+import { getTokensSubscribedToTitle } from "~/models/token";
+import type { Env } from "~/types/env";
+import type { EpisodesResponseSchema } from "~/types/episode";
+import type { FetchUrlResponse } from "~/types/episode/fetch-url-response";
+import {
+  AniListIdSchema,
+  EpisodeNumberSchema,
+  ErrorResponse,
+  SuccessResponse,
+} from "~/types/schema";
+
+const app = new Hono();
+
+app.post(
+  "/",
+  zValidator(
+    "json",
+    z.object({
+      aniListId: AniListIdSchema,
+      episodeNumber: EpisodeNumberSchema,
+    }),
+  ),
+  async (c) => {
+    const { aniListId, episodeNumber } = await c.req.json<{
+      aniListId: number;
+      episodeNumber: number;
+    }>();
+
+    if (!(await verifyQstashHeader(env<Env, typeof c>(c, "workerd"), c.req))) {
+      return c.json(ErrorResponse, { status: 401 });
+    }
+
+    const domain = c.req.url.replace(c.req.path, "");
+
+    console.log(`${domain}/episodes/${aniListId}`);
+    const { success, result: fetchEpisodesResult } = await fetch(
+      `${domain}/episodes/${aniListId}`,
+    ).then((res) => res.json<EpisodesResponseSchema>());
+    if (!success) {
+      return c.json(ErrorResponse, { status: 500 });
+    }
+
+    const { episodes, providerId } = fetchEpisodesResult;
+    const episode = episodes.find(
+      (episode) => episode.number === episodeNumber,
+    );
+    if (!episode) {
+      return c.json(ErrorResponse, { status: 404 });
+    }
+
+    const { success: fetchUrlSuccess, result: fetchUrlResult } = await fetch(
+      `${domain}/episodes/${aniListId}/url`,
+      {
+        method: "POST",
+        headers: {
+          "Content-Type": "application/json",
+        },
+        body: JSON.stringify({
+          id: episode.id,
+          provider: providerId,
+        }),
+      },
+    ).then((res) => res.json<FetchUrlResponse>());
+    if (!fetchUrlSuccess) {
+      return c.json(ErrorResponse, { status: 500 });
+    }
+
+    const tokens = await getTokensSubscribedToTitle(
+      env<Env, typeof c>(c, "workerd"),
+      aniListId,
+    );
+
+    await Promise.all(
+      tokens.map(async (token) => {
+        return sendFcmMessage(
+          mapKeys(
+            readEnvVariable<AdminSdkCredentials>(c.env, "ADMIN_SDK_JSON"),
+            (_, key) => changeStringCase(key, Case.snake_case, Case.camelCase),
+          ) as unknown as AdminSdkCredentials,
+          {
+            token,
+            data: {
+              type: "new_episode",
+              episodes: JSON.stringify(episodes),
+              episodeStreamInfo: JSON.stringify(fetchUrlResult),
+              aniListId: aniListId.toString(),
+              episodeNumber: episodeNumber.toString(),
+            },
+            android: { priority: "high" },
+          },
+        );
+      }),
+    );
+
+    return c.json(SuccessResponse, 200);
+  },
+);
+
+export default app;

src/controllers/upcoming-titles/anilist.ts

@@ -52,7 +52,6 @@ export async function getUpcomingTitlesFromAnilist(env: Env) {
     env,
     "schedule_last_checked_at",
   ).then((value) => (value ? Number(value) : DateTime.now().toUnixInteger()));
-  console.log(lastCheckedScheduleAt);
   const twoDaysFromNow = DateTime.now().plus({ days: 2 }).toUnixInteger();
 
   let currentPage = 1;

src/index.ts

@@ -39,6 +39,12 @@ app.route(
     (controller) => controller.default,
   ),
 );
+app.route(
+  "/new-episode",
+  await import("~/controllers/new-episode").then(
+    (controller) => controller.default,
+  ),
+);
 
 // The OpenAPI documentation will be available at /doc
 app.doc("/openapi.json", {

src/models/token.ts

@@ -1,10 +1,10 @@
-import { eq, or, sql } from "drizzle-orm";
+import { and, eq, gt, or, sql } from "drizzle-orm";
 
 import { TokenAlreadyExistsError } from "~/libs/errors/TokenAlreadyExists";
 import type { Env } from "~/types/env";
 
 import { getDb } from "./db";
-import { deviceTokensTable } from "./schema";
+import { deviceTokensTable, watchStatusTable } from "./schema";
 
 export function saveToken(
   env: Env,
@@ -74,3 +74,24 @@ export function updateDeviceLastConnectedAt(env: Env, deviceId: string) {
     .where(eq(deviceTokensTable.deviceId, deviceId))
     .run();
 }
+
+export function getTokensSubscribedToTitle(env: Env, titleId: number) {
+  return getDb(env)
+    .select({ token: deviceTokensTable.token })
+    .from(deviceTokensTable)
+    .fullJoin(
+      watchStatusTable,
+      eq(deviceTokensTable.deviceId, watchStatusTable.deviceId),
+    )
+    .where(
+      and(
+        eq(watchStatusTable.titleId, titleId),
+        gt(deviceTokensTable.lastConnectedAt, sql`date('now', '-1 month')`),
+      ),
+    )
+    .then((tokens) =>
+      tokens
+        .map(({ token }) => token)
+        .filter((token): token is string => !!token),
+    );
+}

src/types/episode/fetch-url-response.ts

@@ -0,0 +1,18 @@
+import { z } from "zod";
+
+import { SkippableSchema, SuccessResponseSchema } from "~/types/schema";
+
+export type FetchUrlResponseSchema = z.infer<typeof FetchUrlResponseSchema>;
+export const FetchUrlResponseSchema = z.object({
+  source: z.string(),
+  subtitles: z.array(z.object({ url: z.string(), lang: z.string() })),
+  audio: z.array(z.object({ url: z.string(), lang: z.string() })),
+  intro: SkippableSchema,
+  outro: SkippableSchema,
+  headers: z.record(z.string()).optional(),
+});
+
+export type FetchUrlResponse = z.infer<typeof FetchUrlResponse> & {
+  result: FetchUrlResponseSchema;
+};
+export const FetchUrlResponse = SuccessResponseSchema(FetchUrlResponseSchema);

src/types/episode/index.ts

@@ -1,6 +1,6 @@
 import { z } from "zod";
 
-import { EpisodeNumberSchema } from "~/types/schema";
+import { EpisodeNumberSchema, SuccessResponseSchema } from "~/types/schema";
 
 export type Episode = z.infer<typeof Episode>;
 export const Episode = z.object({
@@ -18,3 +18,8 @@ export const EpisodesResponse = z.object({
   providerId: z.string(),
   episodes: z.array(Episode),
 });
+
+export type EpisodesResponseSchema = z.infer<typeof EpisodesResponseSchema> & {
+  result: EpisodesResponse;
+};
+export const EpisodesResponseSchema = SuccessResponseSchema(EpisodesResponse);