feat(twitch): add new live streams trigger

packages/backend/src/apps/twitch/auth/index.ts

@@ -2,6 +2,7 @@ import generateAuthUrl from './generate-auth-url';
 import verifyCredentials from './verify-credentials';
 import refreshToken from './refresh-token';
 import isStillVerified from './is-still-verified';
+import verifyWebhook from './verify-webhook';
 
 export default {
   fields: [
@@ -45,4 +46,5 @@ export default {
   verifyCredentials,
   isStillVerified,
   refreshToken,
+  verifyWebhook,
 };

packages/backend/src/apps/twitch/auth/verify-webhook.ts

@@ -0,0 +1,35 @@
+import crypto from 'crypto';
+import { IGlobalVariable } from '@automatisch/types';
+import appConfig from '../../../config/app';
+
+const verifyWebhook = async ($: IGlobalVariable) => {
+  const signature = $.request.headers[
+    'twitch-eventsub-message-signature'
+  ] as string;
+  const twitchMessageId = $.request.headers[
+    'twitch-eventsub-message-id'
+  ] as string;
+  const twitchMessageTimestamp = $.request.headers[
+    'twitch-eventsub-message-timestamp'
+  ] as string;
+  const rawBody = $.request.rawBody.toString();
+  const hmacMessage = twitchMessageId + twitchMessageTimestamp + rawBody;
+  const hash = crypto
+    .createHmac('sha256', appConfig.webhookSecretKey)
+    .update(hmacMessage)
+    .digest('hex');
+  const hmac = `sha256=${hash}`;
+
+  const isValid = verifySignature(signature, hmac);
+
+  return isValid;
+};
+
+const verifySignature = function (receivedSignature: string, payload: string) {
+  return crypto.timingSafeEqual(
+    Buffer.from(payload),
+    Buffer.from(receivedSignature)
+  );
+};
+
+export default verifyWebhook;

packages/backend/src/apps/twitch/common/auth-scope.ts

@@ -1,3 +1,3 @@
-const authScope: string[] = ['user:read:email'];
+const authScope: string[] = ['user:read:email', 'user:read:follows'];
 
 export default authScope;

packages/backend/src/apps/twitch/dynamic-data/index.ts

@@ -0,0 +1,3 @@
+import listBroadcasters from './list-broadcasters';
+
+export default [listBroadcasters];

packages/backend/src/apps/twitch/dynamic-data/list-broadcasters/index.ts

@@ -0,0 +1,33 @@
+import { IGlobalVariable, IJSONObject } from '@automatisch/types';
+import getCurrentUser from '../../common/get-current-user';
+
+export default {
+  name: 'List broadcasters',
+  key: 'listBroadcasters',
+
+  async run($: IGlobalVariable) {
+    const Broadcasters: {
+      data: IJSONObject[];
+    } = {
+      data: [],
+    };
+    const currentUser = await getCurrentUser($);
+
+    const params = {
+      user_id: currentUser.id,
+    };
+
+    const { data } = await $.http.get('/helix/channels/followed', { params });
+
+    if (data.data?.length) {
+      for (const broadcaster of data.data) {
+        Broadcasters.data.push({
+          value: broadcaster.broadcaster_id,
+          name: broadcaster.broadcaster_name,
+        });
+      }
+    }
+
+    return Broadcasters;
+  },
+};

packages/backend/src/apps/twitch/index.ts

@@ -1,6 +1,8 @@
 import defineApp from '../../helpers/define-app';
 import addAuthHeader from './common/add-auth-header';
 import auth from './auth';
+import triggers from './triggers';
+import dynamicData from './dynamic-data';
 
 export default defineApp({
   name: 'Twitch',
@@ -13,4 +15,6 @@ export default defineApp({
   supportsConnections: true,
   beforeRequest: [addAuthHeader],
   auth,
+  triggers,
+  dynamicData,
 });

packages/backend/src/apps/twitch/triggers/index.ts

@@ -0,0 +1,3 @@
+import newLiveStreams from './new-live-streams';
+
+export default [newLiveStreams];

packages/backend/src/apps/twitch/triggers/new-live-streams/index.ts

@@ -0,0 +1,108 @@
+import Crypto from 'crypto';
+import isEmpty from 'lodash/isEmpty';
+import defineTrigger from '../../../../helpers/define-trigger';
+import appConfig from '../../../../config/app';
+
+type Response = {
+  data: {
+    data: [
+      {
+        id: string;
+      }
+    ];
+  };
+};
+
+export default defineTrigger({
+  name: 'New live streams',
+  key: 'newLiveStreams',
+  type: 'webhook',
+  description:
+    'Triggers when a new live stream starts, regardless of the specific game or language it involves. To include a streamer you are not currently following, input the username of the streamer you want to add.',
+  arguments: [
+    {
+      label: 'Broadcaster',
+      key: 'broadcasterId',
+      type: 'dropdown' as const,
+      required: true,
+      description: '',
+      variables: true,
+      source: {
+        type: 'query',
+        name: 'getDynamicData',
+        arguments: [
+          {
+            name: 'key',
+            value: 'listBroadcasters',
+          },
+        ],
+      },
+    },
+  ],
+
+  async run($) {
+    const dataItem = {
+      raw: $.request.body,
+      meta: {
+        internalId: Crypto.randomUUID(),
+      },
+    };
+
+    $.pushTriggerItem(dataItem);
+  },
+
+  async testRun($) {
+    const lastExecutionStep = await $.getLastExecutionStep();
+
+    if (!isEmpty(lastExecutionStep?.dataOut)) {
+      $.pushTriggerItem({
+        raw: lastExecutionStep.dataOut,
+        meta: {
+          internalId: '',
+        },
+      });
+    }
+  },
+
+  async registerHook($) {
+    const broadcasterId = $.step.parameters.broadcasterId as string;
+
+    const payload = {
+      type: 'stream.online',
+      version: '1',
+      condition: {
+        broadcaster_user_id: broadcasterId,
+      },
+      transport: {
+        method: 'webhook',
+        callback: $.webhookUrl,
+        secret: appConfig.webhookSecretKey,
+      },
+    };
+
+    const response: Response = await $.http.post(
+      '/helix/eventsub/subscriptions',
+      payload,
+      {
+        additionalProperties: {
+          appAccessToken: true,
+        },
+      }
+    );
+
+    await $.flow.setRemoteWebhookId(response.data.data[0].id);
+  },
+
+  async unregisterHook($) {
+    const params = {
+      id: $.flow.remoteWebhookId,
+    };
+
+    await $.http.delete('/helix/eventsub/subscriptions', {
+      params,
+      additionalProperties: {
+        appAccessToken: true,
+      },
+    });
+  },
+});