Lumi/src/services/youtube.js

const { getSetting, setSetting } = require("./settings");
const { incrementMessages } = require("./stats");
const { ensureUserForIdentity } = require("./users");

let client = null;
let pollTimer = null;

async function startYouTubeBot({ commandRouter } = {}) {
  const clientId = getSetting("youtube_client_id");
  const clientSecret = getSetting("youtube_client_secret");
  const refreshToken = getSetting("youtube_bot_refresh_token");
  if (!clientId || !clientSecret || !refreshToken) {
    return null;
  }

  const state = {
    clientId,
    clientSecret,
    refreshToken,
    accessToken: null,
    accessTokenExpiresAt: 0,
    channelId: getSetting("youtube_bot_channel_id", "") || null,
    channelName: null,
    channelAvatar: null,
    liveChatId: null,
    nextPageToken: null,
    skipHistory: true,
    stopped: false,
    commandRouter
  };
  state.sendMessage = async (content) => {
    const liveChatId = await ensureLiveChatId(state);
    if (!liveChatId) {
      throw new Error("No active YouTube live chat is available.");
    }
    return sendChatMessage(state, liveChatId, content);
  };

  client = state;

  try {
    await hydrateBotChannel(state);
  } catch (error) {
    console.error("YouTube bot failed to load channel details", error);
  }

  schedulePoll(state, 1000);
  return state;
}

async function stopYouTubeBot() {
  if (pollTimer) {
    clearTimeout(pollTimer);
    pollTimer = null;
  }
  if (client) {
    client.stopped = true;
  }
  client = null;
}

function getClient() {
  return client;
}

function schedulePoll(state, delayMs) {
  if (state.stopped) {
    return;
  }
  if (pollTimer) {
    clearTimeout(pollTimer);
  }
  pollTimer = setTimeout(() => {
    pollTimer = null;
    pollLiveChat(state).catch((error) => {
      console.error("YouTube chat poll failed", error);
      schedulePoll(state, 10000);
    });
  }, delayMs);
}

async function pollLiveChat(state) {
  if (state.stopped) {
    return;
  }
  const liveChatId = await ensureLiveChatId(state);
  if (!liveChatId) {
    schedulePoll(state, 30000);
    return;
  }

  const response = await listLiveChatMessages(state, liveChatId, state.nextPageToken);
  state.nextPageToken = response.nextPageToken || state.nextPageToken;
  if (state.skipHistory) {
    state.skipHistory = false;
    schedulePoll(state, response.pollingIntervalMillis || 5000);
    return;
  }

  const items = Array.isArray(response.items) ? response.items : [];
  for (const item of items) {
    await handleChatItem(state, liveChatId, item);
  }

  schedulePoll(state, response.pollingIntervalMillis || 5000);
}

async function handleChatItem(state, liveChatId, item) {
  const snippet = item?.snippet;
  const author = item?.authorDetails;
  if (!snippet || !author) {
    return;
  }
  const messageText = snippet.displayMessage;
  if (!messageText) {
    return;
  }
  if (state.channelId && author.channelId === state.channelId) {
    return;
  }

  const displayName = author.displayName || "YouTube User";
  const avatar = author.profileImageUrl || null;
  const profile = ensureUserForIdentity({
    provider: "youtube",
    providerUserId: author.channelId,
    displayName,
    avatar
  });
  incrementMessages(profile.id);

  if (!state.commandRouter) {
    return;
  }

  await state.commandRouter.handleMessage({
    platform: "youtube",
    raw: messageText,
    user: profile,
    platformUser: {
      id: author.channelId,
      displayName,
      username: displayName
    },
    meta: {
      liveChatId,
      messageId: item.id,
      snippet,
      author
    },
    reply: async (content) => {
      try {
        await sendChatMessage(state, liveChatId, content);
      } catch (error) {
        console.error("YouTube command reply failed", error);
      }
    }
  });
}

async function ensureLiveChatId(state) {
  const liveChatId = state.liveChatId || (await findActiveLiveChatId(state));
  if (liveChatId && liveChatId !== state.liveChatId) {
    state.liveChatId = liveChatId;
    state.nextPageToken = null;
    state.skipHistory = true;
  }
  return state.liveChatId;
}

async function hydrateBotChannel(state) {
  const channel = await fetchMyChannel(state);
  if (!channel) {
    return;
  }
  state.channelId = channel.id;
  state.channelName = channel.snippet?.title || null;
  state.channelAvatar = channel.snippet?.thumbnails?.default?.url || null;
  if (state.channelId) {
    setSetting("youtube_bot_channel_id", state.channelId);
  }
}

async function ensureAccessToken(state) {
  const now = Date.now();
  if (state.accessToken && now < state.accessTokenExpiresAt) {
    return state.accessToken;
  }
  const token = await refreshAccessToken(state);
  state.accessToken = token.access_token;
  const expiresIn = Number(token.expires_in || 0);
  state.accessTokenExpiresAt = now + Math.max(30, expiresIn - 60) * 1000;
  return state.accessToken;
}

async function refreshAccessToken(state) {
  const body = new URLSearchParams({
    client_id: state.clientId,
    client_secret: state.clientSecret,
    refresh_token: state.refreshToken,
    grant_type: "refresh_token"
  });
  const response = await fetch("https://oauth2.googleapis.com/token", {
    method: "POST",
    headers: { "Content-Type": "application/x-www-form-urlencoded" },
    body
  });
  if (!response.ok) {
    throw new Error(`YouTube token refresh failed: ${response.status}`);
  }
  return response.json();
}

async function fetchMyChannel(state) {
  const accessToken = await ensureAccessToken(state);
  const response = await fetch(
    "https://www.googleapis.com/youtube/v3/channels?part=snippet&mine=true",
    { headers: { Authorization: `Bearer ${accessToken}` } }
  );
  if (!response.ok) {
    throw new Error(`YouTube channel fetch failed: ${response.status}`);
  }
  const data = await response.json();
  return data.items?.[0] || null;
}

async function findActiveLiveChatId(state) {
  const accessToken = await ensureAccessToken(state);
  const response = await fetch(
    "https://www.googleapis.com/youtube/v3/liveBroadcasts?part=snippet,status&broadcastStatus=active&broadcastType=all",
    { headers: { Authorization: `Bearer ${accessToken}` } }
  );
  if (!response.ok) {
    return null;
  }
  const data = await response.json();
  const broadcast = data.items?.[0];
  return broadcast?.snippet?.liveChatId || null;
}

async function listLiveChatMessages(state, liveChatId, pageToken) {
  const accessToken = await ensureAccessToken(state);
  const params = new URLSearchParams({
    liveChatId,
    part: "snippet,authorDetails",
    maxResults: "200"
  });
  if (pageToken) {
    params.set("pageToken", pageToken);
  }
  const response = await fetch(
    `https://www.googleapis.com/youtube/v3/liveChatMessages?${params.toString()}`,
    { headers: { Authorization: `Bearer ${accessToken}` } }
  );
  if (!response.ok) {
    throw new Error(`YouTube chat fetch failed: ${response.status}`);
  }
  return response.json();
}

async function sendChatMessage(state, liveChatId, messageText) {
  const accessToken = await ensureAccessToken(state);
  const response = await fetch(
    "https://www.googleapis.com/youtube/v3/liveChatMessages?part=snippet",
    {
      method: "POST",
      headers: {
        Authorization: `Bearer ${accessToken}`,
        "Content-Type": "application/json"
      },
      body: JSON.stringify({
        snippet: {
          liveChatId,
          type: "textMessageEvent",
          textMessageDetails: {
            messageText
          }
        }
      })
    }
  );
  if (!response.ok) {
    throw new Error(`YouTube chat send failed: ${response.status}`);
  }
  return response.json();
}

module.exports = {
  startYouTubeBot,
  stopYouTubeBot,
  getClient
};