✨ Twitter upload integration

packages/app/app/studio/[organization]/destinations/page.tsx

@@ -39,7 +39,7 @@ const Destinations = async ({
       case 'youtube':
         return <SiYoutube color="#FF0000" className={className} />;
       case 'twitter':
-        return <SiX className={className} />;
+        return null;
       default:
         return <LuRadio color="#000" className={className} />;
     }
@@ -71,13 +71,19 @@ const Destinations = async ({
                 <TableRow key={_id}>
                   <TableCell className="flex items-center gap-4">
                     <div className="relative">
-                      <Image
-                        className="rounded-full"
-                        src={thumbnail!}
-                        width={50}
-                        height={50}
-                        alt={type}
-                      />
+                      {type === 'twitter' ? (
+                        <div className="bg-black w-[50px] h-[50px] rounded-full">
+                          <SiX color="#fff" className="w-full h-full p-3" />
+                        </div>
+                      ) : (
+                        <Image
+                          className="rounded-full"
+                          src={thumbnail!}
+                          width={50}
+                          height={50}
+                          alt={type}
+                        />
+                      )}
                       {renderSocialTypeIcon(type)}
                     </div>
 

packages/server/src/dtos/session/upload-session.dto.ts

@@ -1,5 +1,5 @@
 import { IUploadSession } from '@interfaces/upload.session.interface';
-import { IsNotEmpty, IsString } from 'class-validator';
+import { IsNotEmpty, IsOptional, IsString } from 'class-validator';
 
 export class UploadSessionDto implements IUploadSession {
   @IsNotEmpty()
@@ -17,4 +17,8 @@ export class UploadSessionDto implements IUploadSession {
   @IsNotEmpty()
   @IsString()
   type: string;
+
+  @IsOptional()
+  @IsString()
+  text?: string;
 }

packages/server/src/interfaces/upload.session.interface.ts

@@ -4,6 +4,7 @@ export interface IUploadSession {
   socialId: string;
   organizationId: string | Types.ObjectId;
   sessionId: string | Types.ObjectId;
-  token?: string;
+  token?: { key?: string; secret: string };
   type: string;
+  text?: string;
 }

packages/server/src/routes/routes.ts

@@ -1093,8 +1093,15 @@ const models: TsoaRoute.Models = {
       socialId: { dataType: 'string', required: true },
       organizationId: { dataType: 'string', required: true },
       sessionId: { dataType: 'string', required: true },
-      token: { dataType: 'string' },
+      token: {
+        dataType: 'nestedObjectLiteral',
+        nestedProperties: {
+          secret: { dataType: 'string', required: true },
+          key: { dataType: 'string' },
+        },
+      },
       type: { dataType: 'string', required: true },
+      text: { dataType: 'string' },
     },
     additionalProperties: false,
   },

packages/server/src/services/session.service.ts

@@ -238,7 +238,12 @@ export default class SessionService {
       (e) => e.type == data.type && e._id == data.socialId,
     );
     if (data.type == 'youtube') {
-      data.token = await refreshAccessToken(token.refreshToken);
+      data.token = {
+        secret: await refreshAccessToken(token.refreshToken),
+      };
+    }
+    if (data.type == 'twitter') {
+      data.token = { key: token.accessToken, secret: token.refreshToken };
     }
     const queue = 'videos';
     const channel = await (await connection).createChannel();

packages/server/src/swagger/swagger.json

@@ -1934,10 +1934,22 @@
             "type": "string"
           },
           "token": {
-            "type": "string"
+            "properties": {
+              "secret": {
+                "type": "string"
+              },
+              "key": {
+                "type": "string"
+              }
+            },
+            "required": ["secret"],
+            "type": "object"
           },
           "type": {
             "type": "string"
+          },
+          "text": {
+            "type": "string"
           }
         },
         "required": ["socialId", "organizationId", "sessionId", "type"],

packages/video-uploader/package.json

@@ -3,7 +3,7 @@
   "version": "1.0.0",
   "main": "index.js",
   "scripts": {
-    "dev": "node -r esbuild-runner/register ./src/video-queue.ts",
+    "dev": "nodemon -r esbuild-runner/register ./src/video-queue.ts --env=development",
     "build": "tsc --build tsconfig.prod.json && tsc-alias -p tsconfig.prod.json",
     "format": "prettier --write \"src/**/*.ts\"",
     "start": "node ./dist/video-queue.js"
@@ -20,6 +20,7 @@
     "amqplib": "^0.10.4",
     "google-auth-library": "^9.11.0",
     "googleapis": "^140.0.0",
-    "mongodb": "^6.7.0"
+    "mongodb": "^6.7.0",
+    "oauth-1.0a": "^2.2.6"
   }
 }

packages/video-uploader/src/utils/twitter.ts

@@ -0,0 +1,196 @@
+import { logger } from './logger';
+import crypto from 'crypto';
+import OAuth from 'oauth-1.0a';
+import fs from 'fs';
+import fetch from 'node-fetch';
+import { config } from 'dotenv';
+config();
+
+function delay(ms: number): Promise<void> {
+  return new Promise((resolve) => setTimeout(resolve, ms));
+}
+const TWITTER_UPLOAD_URL = 'https://upload.twitter.com/1.1/media/upload.json';
+const TWEET_URL_V2 = 'https://api.twitter.com/2/tweets';
+const oauth = new OAuth({
+  consumer: {
+    key: process.env.TWITTER_OAUTH_KEY,
+    secret: process.env.TWITTER_OAUTH_SECRET,
+  },
+  signature_method: 'HMAC-SHA1',
+  hash_function(baseString, key) {
+    return crypto.createHmac('sha1', key).update(baseString).digest('base64');
+  },
+});
+
+function getAuthHeaders(
+  token: { key: string; secret: string },
+  url: string,
+  method: string,
+  params?: {}
+) {
+  const payload = {
+    url: url,
+    method: method,
+    data: params,
+  };
+  return oauth.toHeader(
+    oauth.authorize(payload, {
+      key: token.key,
+      secret: token.secret,
+    })
+  );
+}
+
+function readChunk(
+  filePath: string,
+  start: number,
+  end: number
+): Promise<Buffer> {
+  return new Promise((resolve, reject) => {
+    const chunks: Buffer[] = [];
+    const stream = fs.createReadStream(filePath, { start, end: end - 1 });
+    stream.on('data', (chunk) => chunks.push(Buffer.from(chunk)));
+    stream.on('end', () => resolve(Buffer.concat(chunks)));
+    stream.on('error', reject);
+  });
+}
+
+async function initializeUpload(
+  filePath: string,
+  token: { key: string; secret: string }
+): Promise<string> {
+  const params = {
+    command: 'INIT',
+    total_bytes: fs.statSync(filePath).size.toString(),
+    media_type: 'video/mp4',
+  };
+  const body = new URLSearchParams(params);
+  const authHeader = getAuthHeaders(token, TWITTER_UPLOAD_URL, 'POST', params);
+  const response = await fetch(TWITTER_UPLOAD_URL, {
+    method: 'POST',
+    body: body,
+    headers: {
+      ...authHeader,
+      'content-type': 'application/x-www-form-urlencoded',
+    },
+  });
+  const data = await response.json();
+  return data.media_id_string;
+}
+
+async function uploadVideoChunks(
+  filePath: string,
+  mediaId: string,
+  fileSize: number,
+  token: { key: string; secret: string }
+) {
+  const chunkSize = 5 * 1024 * 1024; // 5 MB in bytes
+  let segmentIndex = 0;
+
+  for (let start = 0; start < fileSize; start += chunkSize, segmentIndex++) {
+    const end = Math.min(start + chunkSize, fileSize);
+    const chunk = await readChunk(filePath, start, end);
+    const base64Chunk = chunk.toString('base64');
+    const params = {
+      command: 'APPEND',
+      media_id: mediaId,
+      media_data: base64Chunk,
+      segment_index: segmentIndex.toString(),
+    };
+    const queryParams = new URLSearchParams(params);
+    const headers = getAuthHeaders(token, TWITTER_UPLOAD_URL, 'POST', params);
+    const response = await fetch(TWITTER_UPLOAD_URL, {
+      method: 'POST',
+      body: queryParams,
+      headers: {
+        ...headers,
+      },
+    });
+    if (!response.ok) {
+      throw new Error(`HTTP error! status: ${response.status}`);
+    }
+  }
+}
+
+async function finalizeUpload(
+  mediaId: string,
+  token: { key: string; secret: string }
+) {
+  let status = 'pending';
+  do {
+    const params = {
+      command: 'FINALIZE',
+      media_id: mediaId,
+    };
+    const body = new URLSearchParams(params);
+    const authHeader = getAuthHeaders(
+      token,
+      TWITTER_UPLOAD_URL,
+      'POST',
+      params
+    );
+    const response = await fetch(TWITTER_UPLOAD_URL, {
+      method: 'POST',
+      body: body,
+      headers: {
+        ...authHeader,
+      },
+    });
+    const data = await response.json();
+    status = data.processing_info.state;
+    if (status === 'succeeded') {
+      logger.info('Media processing succedded');
+    } else if (status === 'failed') {
+      throw new Error('Media processing failed');
+    } else {
+      logger.info(
+        'Media processing is still pending, waiting for completion...'
+      );
+      await delay(data.processing_info.check_after_secs * 1000);
+    }
+  } while (status !== 'succeeded');
+}
+
+async function tweetWithMediaV2(
+  mediaId: string,
+  text: string,
+  token: { key: string; secret: string }
+) {
+  try {
+    const authHeader = getAuthHeaders(token, TWEET_URL_V2, 'POST');
+    const response = await fetch(TWEET_URL_V2, {
+      method: 'POST',
+      headers: {
+        ...authHeader,
+        'Content-Type': 'application/json',
+      },
+      body: JSON.stringify({
+        text: text,
+        media: {
+          media_ids: [mediaId],
+        },
+      }),
+    });
+    const data = await response.json();
+  } catch (e) {
+    logger.error(e.message);
+  }
+}
+
+export async function uploadToTwitter(
+  session: {
+    name: string;
+    description: string;
+    slug: string;
+    published: boolean;
+    coverImage: string;
+  },
+  videoFilePath: string,
+  token: { key: string; secret: string }
+): Promise<void> {
+  const fileSize = fs.statSync(videoFilePath).size;
+  const mediaId = await initializeUpload(videoFilePath, token);
+  await uploadVideoChunks(videoFilePath, mediaId, fileSize, token);
+  await finalizeUpload(mediaId, token);
+  await tweetWithMediaV2(mediaId, session.name, token);
+}