fix(federation): file worker UPDATEs existing attachment rows + broadcasts
The file worker now UPDATEs the attachment row created by processCreateEvent (swapping sourceUrl interim filename to local path) instead of INSERTing a duplicate. Falls back to INSERT for legacy queue entries. After download, broadcasts dm_message_updated so clients see the attachment transition from remote hotlink to local file in real-time.
This commit is contained in:
@@ -5,6 +5,8 @@ import { config } from '../config.js';
|
|||||||
import { isFederationRelayEnabled } from './federationOutbox.js';
|
import { isFederationRelayEnabled } from './federationOutbox.js';
|
||||||
import { buildFederationHeaders } from './federationAuth.js';
|
import { buildFederationHeaders } from './federationAuth.js';
|
||||||
import { generateSnowflake } from './snowflake.js';
|
import { generateSnowflake } from './snowflake.js';
|
||||||
|
import { getDmMessageWithUser } from '../routes/dm.js';
|
||||||
|
import { connectionManager } from '../ws/handler.js';
|
||||||
import type { FederationRelayRequest, FederationRelayResponse, FederationRelayEvent } from '@backspace/shared';
|
import type { FederationRelayRequest, FederationRelayResponse, FederationRelayEvent } from '@backspace/shared';
|
||||||
import fs from 'node:fs';
|
import fs from 'node:fs';
|
||||||
import path from 'node:path';
|
import path from 'node:path';
|
||||||
@@ -442,22 +444,39 @@ async function processFileQueueEntry(
|
|||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
|
||||||
// Create a local attachment record
|
// Update the existing attachment row (created by processCreateEvent with
|
||||||
const attachmentId = generateSnowflake();
|
// sourceUrl as interim filename) to point to the local file
|
||||||
db.insert(schema.attachments)
|
const updated = db.update(schema.attachments)
|
||||||
.values({
|
.set({
|
||||||
id: attachmentId,
|
|
||||||
dmMessageId: entry.dmMessageId,
|
|
||||||
uploaderId: null,
|
|
||||||
filename: localFilename,
|
filename: localFilename,
|
||||||
originalName: entry.originalName,
|
|
||||||
mimetype: entry.mimetype,
|
|
||||||
size: stat.size,
|
size: stat.size,
|
||||||
sourceUrl: entry.sourceUrl,
|
|
||||||
createdAt: now,
|
|
||||||
})
|
})
|
||||||
|
.where(
|
||||||
|
and(
|
||||||
|
eq(schema.attachments.dmMessageId, entry.dmMessageId),
|
||||||
|
eq(schema.attachments.sourceUrl, entry.sourceUrl),
|
||||||
|
),
|
||||||
|
)
|
||||||
.run();
|
.run();
|
||||||
|
|
||||||
|
// Fallback: if no existing row was found (e.g., legacy queue entry from
|
||||||
|
// before processCreateEvent created rows), insert a new one
|
||||||
|
if (updated.changes === 0) {
|
||||||
|
db.insert(schema.attachments)
|
||||||
|
.values({
|
||||||
|
id: generateSnowflake(),
|
||||||
|
dmMessageId: entry.dmMessageId,
|
||||||
|
uploaderId: null,
|
||||||
|
filename: localFilename,
|
||||||
|
originalName: entry.originalName,
|
||||||
|
mimetype: entry.mimetype,
|
||||||
|
size: stat.size,
|
||||||
|
sourceUrl: entry.sourceUrl,
|
||||||
|
createdAt: now,
|
||||||
|
})
|
||||||
|
.run();
|
||||||
|
}
|
||||||
|
|
||||||
// Mark file queue entry as completed
|
// Mark file queue entry as completed
|
||||||
db.update(schema.federationFileQueue)
|
db.update(schema.federationFileQueue)
|
||||||
.set({
|
.set({
|
||||||
@@ -470,6 +489,15 @@ async function processFileQueueEntry(
|
|||||||
console.log(
|
console.log(
|
||||||
`[federation-worker] Downloaded federated file: ${entry.originalName} -> ${localFilename}`,
|
`[federation-worker] Downloaded federated file: ${entry.originalName} -> ${localFilename}`,
|
||||||
);
|
);
|
||||||
|
|
||||||
|
// Notify connected clients that the attachment is now available locally
|
||||||
|
const updatedMsg = getDmMessageWithUser(entry.dmMessageId);
|
||||||
|
if (updatedMsg) {
|
||||||
|
connectionManager.sendToDmMembers(updatedMsg.dmChannelId, {
|
||||||
|
type: 'dm_message_updated',
|
||||||
|
message: updatedMsg,
|
||||||
|
});
|
||||||
|
}
|
||||||
} catch (err) {
|
} catch (err) {
|
||||||
if (err instanceof DOMException && err.name === 'AbortError') {
|
if (err instanceof DOMException && err.name === 'AbortError') {
|
||||||
// Worker is stopping — leave entry as pending for next tick
|
// Worker is stopping — leave entry as pending for next tick
|
||||||
|
|||||||
Reference in New Issue
Block a user