sr2/link-stack
4
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Bridge worker updates

Browse source
This commit is contained in:
Darren Clarke 2024-04-21 16:59:50 +02:00
parent a445762a37
commit f93c4ad317
33 changed files with 17584 additions and 161 deletions
Download patch file Download diff file Expand all files Collapse all files

18
apps/bridge-frontend/app/_lib/database.ts
View file

@ -1,8 +1,20 @@
import { PostgresDialect } from "kysely";
import { PostgresDialect, CamelCasePlugin } from "kysely";
import type { GeneratedAlways } from "kysely";
import { Pool } from "pg";
import { KyselyAuth } from "@auth/kysely-adapter";
import { CamelCasePlugin } from "kysely";
import type { GeneratedAlways } from "kysely";
type GraphileJob = {
taskIdentifier: string;
payload: Record<string, any>;
priority: number;
maxAttempts: number;
key: string;
queueName: string;
};
export const addGraphileJob = async (jobInfo: GraphileJob) => {
await db().insertInto("graphile_worker.jobs").values(jobInfo).execute();
};
interface Database {
User: {

1
apps/bridge-worker/.dockerignore
View file

@ -8,6 +8,5 @@
**/.env*
**/coverage
**/.next
**/amigo.*.json
**/cypress/videos
**/cypress/screenshots

1
apps/bridge-worker/.eslintrc.js
View file

@ -1 +0,0 @@
require("../.eslintrc.js");

32
apps/bridge-worker/Dockerfile Normal file
View file

@ -0,0 +1,32 @@
FROM node:18-alpine as base
FROM base AS builder
ARG APP_DIR=/opt/bridge
RUN mkdir -p ${APP_DIR}/
RUN npm i -g turbo
WORKDIR ${APP_DIR}
COPY . .
RUN turbo prune --scope=bridge-worker --docker
FROM base AS installer
ARG APP_DIR=/opt/bridge
WORKDIR ${APP_DIR}
COPY .gitignore .gitignore
COPY --from=builder ${APP_DIR}/out/json/ .
COPY --from=builder ${APP_DIR}/out/package-lock.json ./package-lock.json
COPY --from=builder ${APP_DIR}/out/full/ .
RUN npm ci
RUN npm i -g turbo
RUN turbo run build --filter=bridge-worker
FROM graphile/worker:0.16.5 as runner
ARG APP_DIR=/opt/bridge
RUN mkdir -p ${APP_DIR}/
ARG BUILD_DIR=${APP_DIR}/apps/bridge-worker/build/main
RUN mkdir -p ${APP_DIR}/
WORKDIR /worker
COPY --from=installer ${BUILD_DIR}/lib ${APP_DIR}/lib
COPY --from=installer ${BUILD_DIR}/tasks ${APP_DIR}/tasks
COPY --from=installer ${APP_DIR}/apps/bridge-worker/graphile.config.prod.js ./graphile.config.js
COPY --from=installer ${APP_DIR}/node_modules ${APP_DIR}/node_modules
COPY --from=installer ${APP_DIR}/package.json ${APP_DIR}/package.json

3
apps/bridge-worker/babel.config.json
View file

@ -1,3 +0,0 @@
{
"presets": ["@digiresilience/babel-preset-bridge"]
}

10
apps/bridge-worker/graphile.config.prod.js Normal file
View file

@ -0,0 +1,10 @@
module.exports = {
worker: {
connectionString: process.env.DATABASE_URL,
maxPoolSize: 10,
pollInterval: 2000,
concurrentJobs: 3,
taskDirectory: "/opt/bridge/tasks",
fileExtensions: [".js", ".cjs", ".mjs"],
},
};

13
apps/bridge-worker/graphile.config.ts Normal file
View file

@ -0,0 +1,13 @@
import type {} from "graphile-config";
import type {} from "graphile-worker";
const preset: GraphileConfig.Preset = {
worker: {
connectionString: process.env.DATABASE_URL,
maxPoolSize: 10,
pollInterval: 2000,
fileExtensions: [".ts"],
},
};
export default preset;

73
apps/bridge-worker/index.ts
View file

@ -1,10 +1,11 @@
import * as Worker from "graphile-worker";
import { parseCronItems } from "graphile-worker";
import { initPgp } from "./db.js";
import logger from "./logger.js";
import workerUtils from "./utils.js";
import { assertFfmpegAvailable } from "./lib/media-convert.js";
import { run, Runner, RunnerOptions } from "graphile-worker";
// import { parseCronItems } from "graphile-worker";
// import { initPgp } from "./db.js";
// import logger from "./logger.js";
// import workerUtils from "./utils.js";
// import { assertFfmpegAvailable } from "./lib/media-convert.js";
/*
const logFactory = (scope: any) => (level: any, message: any, meta: any) => {
const pinoLevel = level === "warning" ? "warn" : level;
const childLogger = logger.child({ scope });
@ -12,40 +13,28 @@ const logFactory = (scope: any) => (level: any, message: any, meta: any) => {
else childLogger[pinoLevel](message);
};
export const configWorker = async (): Promise<Worker.RunnerOptions> => {
const {
connection,
concurrency,
pollInterval,
leafcutter: {
enabled: leafcutterEnabled
}
} = config.worker;
logger.info({ concurrency, pollInterval }, "Starting worker");
*/
const getConfig = (): RunnerOptions => {
// logger.info({ concurrency, pollInterval }, "Starting worker");
return {
concurrency,
pollInterval,
logger: new Worker.Logger(logFactory),
connectionString: connection,
concurrency: 10,
pollInterval: 1000,
// logger: new Worker.Logger(logFactory),
connectionString: process.env.DATABASE_URL,
// eslint-disable-next-line unicorn/prefer-module
taskDirectory: `${__dirname}/tasks`,
parsedCronItems: parseCronItems(leafcutterEnabled ?
[{ task: "import-label-studio", pattern: "*/15 * * * *" },
{ task: "import-leafcutter", pattern: "*/17 * * * *" }] :
[])
// parsedCronItems: parseCronItems(
// leafcutterEnabled
// ? [
// { task: "import-label-studio", pattern: "*/15 * * * *" },
// { task: "import-leafcutter", pattern: "*/17 * * * *" },
// ]
// : [],
// )
};
};
export const startWorker = async (): Promise<Worker.Runner> => {
// ensure ffmpeg is installed and working
await assertFfmpegAvailable();
logger.info("ffmpeg found");
await workerUtils.migrate();
logger.info("worker database migrated");
initPgp();
export const startWorker = async (): Promise<Runner> => {
const workerConfig = await configWorker();
const worker = await Worker.run(workerConfig);
return worker;
@ -55,9 +44,15 @@ export const stopWorker = async (): Promise<void> => {
await worker.stop();
};
const worker = defState("worker", {
start: startWorker,
stop: stopWorker,
});
const main = async () => {
await worker.start();
process.on("SIGTERM", async () => {
await worker.stop();
process.exit(0);
});
};
export default worker;
main().catch((err) => {
console.error(err);
process.exit(1);
});

4
apps/bridge-worker/common.ts → apps/bridge-worker/lib/common.ts
View file

@ -1,9 +1,11 @@
/* eslint-disable camelcase */
import { SavedVoiceProvider } from "@digiresilience/bridge-db";
// import { SavedVoiceProvider } from "@digiresilience/bridge-db";
import Twilio from "twilio";
import { CallInstance } from "twilio/lib/rest/api/v2010/account/call";
import { Zammad, getOrCreateUser } from "./zammad";
type SavedVoiceProvider = any;
export const twilioClientFor = (
provider: SavedVoiceProvider,
): Twilio.Twilio => {

11
apps/bridge-worker/db.ts → apps/bridge-worker/lib/db.ts
View file

@ -1,3 +1,4 @@
/*
import pgPromise from "pg-promise";
import * as pgMonitor from "pg-monitor";
import {
@ -37,15 +38,19 @@ const initDb = (): AppDatabase => {
export const stopDb = async (db: AppDatabase): Promise<void> => {
return db.$pool.end();
};
*/
export type AppDatabase = any;
export const withDb = <T>(f: (db: AppDatabase) => Promise<T>): Promise<T> => {
const db = initDb();
/*
const db = initDb();
initDiagnostics(config.logging.sql, pgpInitOptions);
try {
return f(db);
} finally {
stopDiagnostics();
}
*/
return f(null);
};
export type { AppDatabase } from "@digiresilience/bridge-db";

0
apps/bridge-worker/lib/facebook.ts Normal file
View file

11
apps/bridge-worker/lib/logger.ts Normal file
View file

@ -0,0 +1,11 @@
//import { defState } from "@digiresilience/montar";
//import { configureLogger } from "@digiresilience/bridge-common";
// import config from "@digiresilience/bridge-config";
//export const logger = defState("workerLogger", {
// start: async () => configureLogger(config),
//});
//export default logger;
export const logger = {};
export default logger;

9
apps/bridge-worker/utils.ts → apps/bridge-worker/lib/utils.ts
View file

@ -1,7 +1,8 @@
import * as Worker from "graphile-worker";
import { defState } from "@digiresilience/montar";
import config from "@digiresilience/bridge-config";
// import { defState } from "@digiresilience/montar";
//import config from "@digiresilience/bridge-config";
/*
const startWorkerUtils = async (): Promise<Worker.WorkerUtils> => {
const workerUtils = await Worker.makeWorkerUtils({
connectionString: config.worker.connection,
@ -18,4 +19,8 @@ const workerUtils = defState("workerUtils", {
stop: stopWorkerUtils,
});
*/
export const workerUtils: any = {};
export default workerUtils;

0
apps/bridge-worker/lib/voice.ts Normal file
View file

0
apps/bridge-worker/lib/whatsapp.ts Normal file
View file

0
apps/bridge-worker/zammad.ts → apps/bridge-worker/lib/zammad.ts
View file

8
apps/bridge-worker/logger.ts
View file

@ -1,8 +0,0 @@
import { defState } from "@digiresilience/montar";
import { configureLogger } from "@digiresilience/bridge-common";
import config from "@digiresilience/bridge-config";
export const logger = defState("workerLogger", {
start: async () => configureLogger(config),
});
export default logger;

43
apps/bridge-worker/package.json
View file

@ -1,55 +1,34 @@
{
"name": "bridge-worker",
"version": "0.2.0",
"main": "build/main/index.js",
"type": "module",
"author": "Abel Luck <abel@guardianproject.info>",
"author": "Darren Clarke <darren@redaranj.com>",
"license": "AGPL-3.0-or-later",
"scripts": {
"build": "tsc -p tsconfig.json",
"dev": "NODE_OPTIONS=\"--loader ts-node/esm\" graphile-worker"
},
"dependencies": {
"@hapi/wreck": "^18.1.0",
"fluent-ffmpeg": "^2.1.2",
"graphile-worker": "^0.16.5",
"html-to-text": "^9.0.5",
"node-fetch": "^3",
"pg-promise": "^11.6.0",
"jest": "^29.7.0",
"kysely": "^0.27.3",
"pg": "^8.11.5",
"remeda": "^1.60.1",
"twilio": "^5.0.4"
},
"devDependencies": {
"ts-config": "*",
"@babel/core": "7.24.4",
"@babel/preset-env": "7.24.4",
"@babel/preset-typescript": "7.24.1",
"@types/fluent-ffmpeg": "^2.1.24",
"@types/jest": "^29.5.12",
"eslint": "^9.0.0",
"jest": "^29.7.0",
"jest-circus": "^29.7.0",
"jest-junit": "^16.0.0",
"nodemon": "^3.1.0",
"pino-pretty": "^11.0.0",
"prettier": "^3.2.5",
"ts-config": "*",
"ts-node": "^10.9.2",
"typedoc": "^0.25.13",
"typescript": "^5.4.5"
},
"nodemonConfig": {
"ignore": [
"docs/*"
],
"ext": "ts,json,js"
},
"scripts": {
"build-xxx": "tsc -p tsconfig.json",
"build-test": "tsc -p tsconfig.json",
"doc:html": "typedoc src/ --exclude '**/*.test.ts' --exclude '**/*.spec.ts' --name $npm_package_name --readme README.md --target es2019 --mode file --out build/docs",
"doc": "yarn run doc:html",
"fix:lint": "eslint src --ext .ts --fix",
"fix:prettier": "prettier \"src/**/*.ts\" --write",
"test:jest": "JEST_CIRCUS=1 jest --coverage --forceExit --detectOpenHandles --reporters=default --reporters=jest-junit",
"test:jest-verbose": "yarn test:jest --verbose --silent=false",
"test": "yarn test:jest",
"lint": "yarn lint:lint && yarn lint:prettier",
"lint:lint": "eslint src --ext .ts",
"lint:prettier": "prettier \"src/**/*.ts\" --list-different",
"watch:test": "yarn test:jest --watchAll"
}
}

30
apps/bridge-worker/tasks/notify-webhook.ts → apps/bridge-worker/tasks/common/notify-webhook.ts
View file

@ -1,7 +1,7 @@
import Wreck from "@hapi/wreck";
import * as R from "remeda";
import { withDb, AppDatabase } from "../db";
import logger from "../logger";
import { withDb, AppDatabase } from "../../lib/db";
// import logger from "../logger";
export interface WebhookOptions {
webhookId: string;
@ -14,10 +14,10 @@ const notifyWebhooksTask = async (options: WebhookOptions): Promise<void> =>
const webhook = await db.webhooks.findById({ id: webhookId });
if (!webhook) {
logger.debug(
{ webhookId },
"notify-webhook: no webhook registered with id"
);
// logger.debug(
// { webhookId },
// "notify-webhook: no webhook registered with id",
// );
return;
}
@ -28,7 +28,7 @@ const notifyWebhooksTask = async (options: WebhookOptions): Promise<void> =>
acc[h.header] = h.value;
return acc;
},
{}
{},
);
const wreck = Wreck.defaults({
@ -38,18 +38,18 @@ const notifyWebhooksTask = async (options: WebhookOptions): Promise<void> =>
// http errors will bubble up and cause the job to fail and be retried
try {
logger.debug(
{ webhookId, endpointUrl, httpMethod },
"notify-webhook: notifying"
);
// logger.debug(
// { webhookId, endpointUrl, httpMethod },
// "notify-webhook: notifying",
// );
await (httpMethod === "post"
? wreck.post(endpointUrl, { payload })
: wreck.put(endpointUrl, { payload }));
} catch (error: any) {
logger.error(
{ webhookId, error: error.output },
"notify-webhook failed with this error"
);
// logger.error(
// { webhookId, error: error.output },
// "notify-webhook failed with this error",
// );
throw new Error(`webhook failed webhookId=${webhookId}`);
}
});

12
apps/bridge-worker/tasks/facebook/receive-facebook-message.ts Normal file
View file

@ -0,0 +1,12 @@
interface ReceiveFacebookMessageTaskOptions {}
const receiveFacebookMessageTask = async (
options: ReceiveFacebookMessageTaskOptions,
): Promise<void> => {
console.log(options);
// withDb(async (db: AppDatabase) => {
// await notifyWebhooks(db, options);
// });
};
export default receiveFacebookMessageTask;

12
apps/bridge-worker/tasks/facebook/send-facebook-message.ts Normal file
View file

@ -0,0 +1,12 @@
interface SendFacebookMessageTaskOptions {}
const sendFacebookMessageTask = async (
options: SendFacebookMessageTaskOptions,
): Promise<void> => {
console.log(options);
// withDb(async (db: AppDatabase) => {
// await notifyWebhooks(db, options);
// });
};
export default sendFacebookMessageTask;

15
apps/bridge-worker/tasks/import-label-studio.ts → apps/bridge-worker/tasks/leafcutter/import-label-studio.ts
View file

@ -1,10 +1,11 @@
/* eslint-disable camelcase */
import { convert } from "html-to-text";
import fetch from "node-fetch";
import { URLSearchParams } from "url";
import { withDb, AppDatabase } from "../db";
import { loadConfig } from "@digiresilience/bridge-config";
import { tagMap } from "../lib/tag-map";
import { withDb, AppDatabase } from "../../lib/db";
// import { loadConfig } from "@digiresilience/bridge-config";
import { tagMap } from "../../lib/tag-map";
const config: any = {};
type FormattedZammadTicket = {
data: Record<string, unknown>;
@ -17,7 +18,7 @@ const getZammadTickets = async (
): Promise<[boolean, FormattedZammadTicket[]]> => {
const {
leafcutter: { zammadApiUrl, zammadApiKey, contributorName, contributorId },
} = await loadConfig();
} = config;
const headers = { Authorization: `Token ${zammadApiKey}` };
let shouldContinue = false;
const docs = [];
@ -161,7 +162,7 @@ const fetchFromZammad = async (
const sendToLabelStudio = async (tickets: FormattedZammadTicket[]) => {
const {
leafcutter: { labelStudioApiUrl, labelStudioApiKey },
} = await loadConfig();
} = config;
const headers = {
Authorization: `Token ${labelStudioApiKey}`,
@ -185,7 +186,7 @@ const importLabelStudioTask = async (): Promise<void> => {
withDb(async (db: AppDatabase) => {
const {
leafcutter: { contributorName },
} = await loadConfig();
} = config;
const settingName = `${contributorName}ImportLabelStudioTask`;
const res: any = await db.settings.findByName(settingName);
const startTimestamp = res?.value?.minUpdatedTimestamp

14
apps/bridge-worker/tasks/import-leafcutter.ts → apps/bridge-worker/tasks/leafcutter/import-leafcutter.ts
View file

@ -1,8 +1,9 @@
/* eslint-disable camelcase */
import fetch from "node-fetch";
import { URLSearchParams } from "url";
import { withDb, AppDatabase } from "../db";
import { loadConfig } from "@digiresilience/bridge-config";
import { withDb, AppDatabase } from "../../lib/db";
// import { loadConfig } from "@digiresilience/bridge-config";
const config: any = {};
type LabelStudioTicket = {
id: string;
@ -32,8 +33,7 @@ const getLabelStudioTickets = async (
): Promise<LabelStudioTicket[]> => {
const {
leafcutter: { labelStudioApiUrl, labelStudioApiKey },
} = await loadConfig();
} = config;
const headers = {
Authorization: `Token ${labelStudioApiKey}`,
Accept: "application/json",
@ -90,7 +90,7 @@ const sendToLeafcutter = async (tickets: LabelStudioTicket[]) => {
opensearchUsername,
opensearchPassword,
},
} = await loadConfig();
} = config;
console.log({ tickets });
const filteredTickets = tickets.filter((ticket) => ticket.is_labeled);
@ -148,7 +148,7 @@ const importLeafcutterTask = async (): Promise<void> => {
withDb(async (db: AppDatabase) => {
const {
leafcutter: { contributorName },
} = await loadConfig();
} = config;
const settingName = `${contributorName}ImportLeafcutterTask`;
const res: any = await db.settings.findByName(settingName);
const startTimestamp = res?.value?.minUpdatedTimestamp

29
apps/bridge-worker/tasks/signald-message.ts → apps/bridge-worker/tasks/signal/signald-message.ts
View file

@ -1,8 +1,10 @@
/* eslint-disable camelcase */
import logger from "../logger";
import { IncomingMessagev1 } from "@digiresilience/node-signald/build/main/generated";
import { withDb, AppDatabase } from "../db";
import workerUtils from "../utils";
// import logger from "../logger";
// import { IncomingMessagev1 } from "@digiresilience/node-signald/build/main/generated";
import { withDb, AppDatabase } from "../../lib/db";
import workerUtils from "../../lib/utils";
type IncomingMessagev1 = any;
interface WebhookPayload {
to: string;
@ -25,7 +27,8 @@ interface SignaldMessageTaskOptions {
}
const formatPayload = (opts: SignaldMessageTaskOptions): WebhookPayload => {
const { botId, botPhoneNumber, message, attachment, filename, mimetype } = opts;
const { botId, botPhoneNumber, message, attachment, filename, mimetype } =
opts;
const { source, timestamp, data_message: dataMessage } = message;
const { number }: any = source;
@ -46,7 +49,7 @@ const formatPayload = (opts: SignaldMessageTaskOptions): WebhookPayload => {
const notifyWebhooks = async (
db: AppDatabase,
messageInfo: SignaldMessageTaskOptions
messageInfo: SignaldMessageTaskOptions,
) => {
const {
botId,
@ -54,16 +57,16 @@ const notifyWebhooks = async (
} = messageInfo;
const webhooks = await db.webhooks.findAllByBackendId("signal", botId);
if (webhooks && webhooks.length === 0) {
logger.debug({ botId }, "no webhooks registered for signal bot");
// logger.debug({ botId }, "no webhooks registered for signal bot");
return;
}
webhooks.forEach(({ id }) => {
const payload = formatPayload(messageInfo);
logger.debug(
{ payload },
"formatted signal bot payload for notify-webhook"
);
// logger.debug(
// { payload },
// "formatted signal bot payload for notify-webhook",
// );
workerUtils.addJob(
"notify-webhook",
{
@ -73,13 +76,13 @@ const notifyWebhooks = async (
{
// this de-deduplicates the job
jobKey: `webhook-${id}-message-${botId}-${timestamp}`,
}
},
);
});
};
const signaldMessageTask = async (
options: SignaldMessageTaskOptions
options: SignaldMessageTaskOptions,
): Promise<void> => {
console.log(options);
withDb(async (db: AppDatabase) => {

14
apps/bridge-worker/tasks/twilio-recording.ts → apps/bridge-worker/tasks/voice/twilio-recording.ts
View file

@ -1,8 +1,8 @@
import Wreck from "@hapi/wreck";
import { withDb, AppDatabase } from "../db";
import { twilioClientFor } from "../common";
import { withDb, AppDatabase } from "../../lib/db";
import { twilioClientFor } from "../../lib/common";
import { CallInstance } from "twilio/lib/rest/api/v2010/account/call";
import workerUtils from "../utils";
import workerUtils from "../../lib/utils";
interface WebhookPayload {
startTime: string;
@ -27,7 +27,7 @@ const getTwilioRecording = async (url: string) => {
const formatPayload = (
call: CallInstance,
recording: Buffer
recording: Buffer,
): WebhookPayload => {
return {
startTime: call.startTime.toISOString(),
@ -45,7 +45,7 @@ const notifyWebhooks = async (
db: AppDatabase,
voiceLineId: string,
call: CallInstance,
recording: Buffer
recording: Buffer,
) => {
const webhooks = await db.webhooks.findAllByBackendId("voice", voiceLineId);
if (webhooks && webhooks.length === 0) return;
@ -61,7 +61,7 @@ const notifyWebhooks = async (
{
// this de-depuplicates the job
jobKey: `webhook-${id}-call-${call.sid}`,
}
},
);
});
};
@ -74,7 +74,7 @@ interface TwilioRecordingTaskOptions {
}
const twilioRecordingTask = async (
options: TwilioRecordingTaskOptions
options: TwilioRecordingTaskOptions,
): Promise<void> =>
withDb(async (db: AppDatabase) => {
const { voiceLineId, accountSid, callSid, recordingSid } = options;

8
apps/bridge-worker/tasks/voice-line-audio-update.ts → apps/bridge-worker/tasks/voice/voice-line-audio-update.ts
View file

@ -1,6 +1,6 @@
import { createHash } from "crypto";
import { withDb, AppDatabase } from "../db";
import { convert } from "../lib/media-convert";
import { withDb, AppDatabase } from "../../lib/db";
import { convert } from "../../lib/media-convert";
interface VoiceLineAudioUpdateTaskOptions {
voiceLineId: string;
@ -13,7 +13,7 @@ const sha1sum = (v: any) => {
};
const voiceLineAudioUpdateTask = async (
payload: VoiceLineAudioUpdateTaskOptions
payload: VoiceLineAudioUpdateTaskOptions,
): Promise<void> =>
withDb(async (db: AppDatabase) => {
const { voiceLineId } = payload;
@ -41,7 +41,7 @@ const voiceLineAudioUpdateTask = async (
"audio/mpeg": mp3.toString("base64"),
checksum: webmSha1,
},
}
},
);
});

6
apps/bridge-worker/tasks/voice-line-delete.ts → apps/bridge-worker/tasks/voice/voice-line-delete.ts
View file

@ -1,6 +1,8 @@
import Twilio from "twilio";
import config from "@digiresilience/bridge-config";
import { withDb, AppDatabase } from "../db";
// import config from "@digiresilience/bridge-config";
import { withDb, AppDatabase } from "../../lib/db";
const config: any = {};
interface VoiceLineDeleteTaskOptions {
voiceLineId: string;

6
apps/bridge-worker/tasks/voice-line-provider-update.ts → apps/bridge-worker/tasks/voice/voice-line-provider-update.ts
View file

@ -1,6 +1,8 @@
import Twilio from "twilio";
import config from "@digiresilience/bridge-config";
import { withDb, AppDatabase } from "../db";
// import config from "@digiresilience/bridge-config";
import { withDb, AppDatabase } from "../../lib/db";
const config: any = {};
interface VoiceLineUpdateTaskOptions {
voiceLineId: string;

19
apps/bridge-worker/tasks/whatsapp-message.ts → apps/bridge-worker/tasks/whatsapp/receive-whatsapp-message.ts
View file

@ -1,6 +1,6 @@
/* eslint-disable camelcase */
import { withDb, AppDatabase } from "../db";
import workerUtils from "../utils";
import { withDb, AppDatabase } from "../../lib/db";
import workerUtils from "../../lib/utils";
interface WebhookPayload {
to: string;
@ -25,7 +25,7 @@ interface WhatsappMessageTaskOptions {
}
const formatPayload = (
messageInfo: WhatsappMessageTaskOptions
messageInfo: WhatsappMessageTaskOptions,
): WebhookPayload => {
const {
waMessageId,
@ -37,8 +37,9 @@ const formatPayload = (
botPhoneNumber,
} = messageInfo;
const parsedMessage = JSON.parse(waMessage);
const message = parsedMessage.message?.conversation ??
parsedMessage.message?.extendedTextMessage?.text ??
const message =
parsedMessage.message?.conversation ??
parsedMessage.message?.extendedTextMessage?.text ??
parsedMessage.message?.imageMessage?.caption ??
parsedMessage.message?.videoMessage?.caption;
@ -56,12 +57,12 @@ const formatPayload = (
const notifyWebhooks = async (
db: AppDatabase,
messageInfo: WhatsappMessageTaskOptions
messageInfo: WhatsappMessageTaskOptions,
) => {
const { waMessageId, whatsappBotId } = messageInfo;
const webhooks = await db.webhooks.findAllByBackendId(
"whatsapp",
whatsappBotId
whatsappBotId,
);
if (webhooks && webhooks.length === 0) return;
@ -77,13 +78,13 @@ const notifyWebhooks = async (
{
// this de-deduplicates the job
jobKey: `webhook-${id}-message-${waMessageId}`,
}
},
);
});
};
const whatsappMessageTask = async (
options: WhatsappMessageTaskOptions
options: WhatsappMessageTaskOptions,
): Promise<void> => {
console.log(options);
withDb(async (db: AppDatabase) => {

8
apps/bridge-worker/tsconfig.json
View file

@ -1,10 +1,16 @@
{
"extends": "tsconfig",
"extends": "ts-config",
"compilerOptions": {
"outDir": "build/main",
"module": "CommonJS",
"esModuleInterop": true,
"skipLibCheck": true
},
"ts-node": {
"module": "ESNext",
"target": "ESNext",
"moduleResolution": "node"
},
"include": ["**/*.ts", "**/.*.ts"],
"exclude": ["node_modules", "build"]
}

67
apps/index.ts Normal file
View file

@ -0,0 +1,67 @@
import { run, Runner, RunnerOptions } from "graphile-worker";
import { parseCronItems } from "graphile-worker";
// import { initPgp } from "./db.js";
// import logger from "./logger.js";
// import workerUtils from "./utils.js";
import { assertFfmpegAvailable } from "./lib/media-convert.js";
/*
const logFactory = (scope: any) => (level: any, message: any, meta: any) => {
const pinoLevel = level === "warning" ? "warn" : level;
const childLogger = logger.child({ scope });
if (meta) childLogger[pinoLevel](meta, message);
else childLogger[pinoLevel](message);
};
*/
const getConfig = (): RunnerOptions => {
// logger.info({ concurrency, pollInterval }, "Starting worker");
return {
concurrency: 10,
pollInterval: 1000,
// logger: new Worker.Logger(logFactory),
connectionString: process.env.DATABASE_URL,
// eslint-disable-next-line unicorn/prefer-module
taskDirectory: `${__dirname}/tasks`,
// parsedCronItems: parseCronItems(
// leafcutterEnabled
// ? [
// { task: "import-label-studio", pattern: "*/15 * * * *" },
// { task: "import-leafcutter", pattern: "*/17 * * * *" },
// ]
// : [],
// )
};
};
export const startWorker = async (): Promise<Runner> => {
// ensure ffmpeg is installed and working
await assertFfmpegAvailable();
// logger.info("ffmpeg found");
// await workerUtils.migrate();
// logger.info("worker database migrated");
// initPgp();
const workerConfig = await configWorker();
const worker = await Worker.run(workerConfig);
return worker;
};
export const stopWorker = async (): Promise<void> => {
await worker.stop();
};
const main = async () => {
await worker.start();
process.on("SIGTERM", async () => {
await worker.stop();
process.exit(0);
});
};
main().catch((err) => {
console.error(err);
process.exit(1);
});