feeddeck/supabase/functions/_cmd/worker/worker.ts

import { connect, Redis } from "redis";
import { createClient, SupabaseClient } from "@supabase/supabase-js";

import { log } from "../../_shared/utils/log.ts";
import { getFeed } from "../../_shared/feed/feed.ts";
import {
  FEEDDECK_REDIS_HOSTNAME,
  FEEDDECK_REDIS_PASSWORD,
  FEEDDECK_REDIS_PORT,
  FEEDDECK_REDIS_USERNAME,
  FEEDDECK_SUPABASE_URL,
} from "../../_shared/utils/constants.ts";
import { FEEDDECK_SUPABASE_SERVICE_ROLE_KEY } from "../../_shared/utils/constants.ts";

/**
 * `runWorker` starts the worker which is responsible for fetching the feeds for all scheduled sources. For this a
 * worker will listen for new "jobs" in the Redis queue and fetch the feed for the source and updates the database
 * entries.
 */
export const runWorker = async () => {
  /**
   * Create a new Supabase client which is used to update all sources and items in the Supabase database.
   */
  const adminSupabaseClient = createClient(
    FEEDDECK_SUPABASE_URL,
    FEEDDECK_SUPABASE_SERVICE_ROLE_KEY,
    {
      auth: {
        autoRefreshToken: false,
        persistSession: false,
      },
    },
  );

  /**
   * Create a new Redis client which is used to listen for new sources which should be fetched and for caching data when
   * needed, e.g. the media urls for the Google News provider.
   */
  const redisClient = await connect({
    hostname: FEEDDECK_REDIS_HOSTNAME,
    port: FEEDDECK_REDIS_PORT,
    username: FEEDDECK_REDIS_USERNAME,
    password: FEEDDECK_REDIS_PASSWORD,
  });

  /**
   * Run the `listenForSources` function in an endless loop.
   */
  while (true) {
    await listenForSources(adminSupabaseClient, redisClient);
  }
};

const listenForSources = async (
  supabaseClient: SupabaseClient,
  redisClient: Redis,
) => {
  try {
    /**
     * Listen for new sources in the Redis queue. Once a valid source is received we get the source and profile from the
     * Redis data.
     */
    const data = await redisClient.blpop(1000 * 60, "sources");
    if (data && data[0] === "sources") {
      const { source: redisSource, profile: redisProfile } = JSON.parse(
        data[1],
      );
      log("info", "Received source", {
        "source": redisSource.id,
        "profile": redisProfile.id,
      });

      /**
       * Fetch the feed for the source using the created Supabase client, Redis client and the source and profile data
       * from Redis.
       */
      const { source, items } = await getFeed(
        supabaseClient,
        redisClient,
        redisProfile,
        redisSource,
      );

      /**
       * Update the source and items in the Supabase database, when the returned list of items contains at least one
       * item. We have to use `upsert` instead of `update` to update do bulk "updates" for all sources and items.
       */
      if (items.length > 0) {
        const { error: sourceError } = await supabaseClient.from("sources")
          .upsert(
            source,
          );
        if (sourceError) {
          log("error", "Failed to save sources", { "error": sourceError });
        }

        const { error: itemsError } = await supabaseClient.from("items")
          .upsert(
            items,
          );
        if (itemsError) {
          log("error", "Failed to save items", { "error": itemsError });
        }

        log("info", "Updated source", {
          "source": redisSource.id,
          "profile": redisProfile.id,
        });
      }
    }
  } catch (err) {
    log("error", "Failed to listen for sources", { "error": err.toString() });
  }
};