server/database/entities/Queue.ts

125 lines
2.7 KiB
TypeScript
Raw Normal View History

2024-04-07 06:16:54 +02:00
import { config } from "config-manager";
2024-04-07 07:30:49 +02:00
// import { Worker } from "bullmq";
import { type StatusWithRelations, statusToLysand } from "./Status";
2023-11-27 02:40:44 +01:00
2024-03-12 19:27:50 +01:00
/* export const federationWorker = new Worker(
2023-11-27 02:40:44 +01:00
"federation",
async job => {
await job.updateProgress(0);
switch (job.name) {
case "federation": {
// eslint-disable-next-line @typescript-eslint/no-unsafe-member-access
const statusId = job.data.id as string;
const status = await client.status.findUnique({
where: { id: statusId },
include: statusAndUserRelations,
});
if (!status) return;
// Only get remote users that follow the author of the status, and the remote mentioned users
const peopleToSendTo = await client.user.findMany({
where: {
OR: [
["public", "unlisted", "private"].includes(
status.visibility
)
? {
relationships: {
some: {
subjectId: status.authorId,
following: true,
},
},
instanceId: {
not: null,
},
}
2023-11-27 02:40:44 +01:00
: {},
// Mentioned users
{
id: {
in: status.mentions.map(m => m.id),
},
instanceId: {
not: null,
},
},
],
},
});
let peopleDone = 0;
// Spawn sendToServer job for each user
for (const person of peopleToSendTo) {
await federationQueue.add("sendToServer", {
id: statusId,
user: person,
});
peopleDone++;
await job.updateProgress(
Math.round((peopleDone / peopleToSendTo.length) * 100)
);
}
break;
}
case "sendToServer": {
// eslint-disable-next-line @typescript-eslint/no-unsafe-member-access
const statusId = job.data.id as string;
// eslint-disable-next-line @typescript-eslint/no-unsafe-member-access
const user = job.data.user as User;
const status = await client.status.findUnique({
where: { id: statusId },
include: statusAndUserRelations,
});
if (!status) return;
const response = await federateStatusTo(
status,
status.author,
user
);
if (response.status !== 200) {
throw new Error(
`Federation error: ${response.status} ${response.statusText}`
);
}
break;
}
}
await job.updateProgress(100);
return true;
},
{
connection: {
host: config.redis.queue.host,
port: config.redis.queue.port,
password: config.redis.queue.password,
2023-11-27 06:30:57 +01:00
db: config.redis.queue.database || undefined,
2023-11-27 02:40:44 +01:00
},
removeOnComplete: {
count: 400,
},
removeOnFail: {
count: 3000,
},
}
2024-03-12 19:27:50 +01:00
); */
2023-11-27 02:40:44 +01:00
export const addStatusFederationJob = async (statusId: string) => {
2024-04-07 07:30:49 +02:00
/* await federationQueue.add("federation", {
2023-11-27 02:40:44 +01:00
id: statusId,
2024-03-12 19:27:50 +01:00
}); */
2023-11-27 02:40:44 +01:00
};