Matiss/Final-Year-Project
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor(backend): simplify media schema and recording metadata

Browse Source
This commit is contained in:
Matiss Jurevics
2026-03-11 17:15:00 +00:00
parent 662d8d7b90
commit c6919d8174
18 changed files with 223 additions and 113 deletions
Download Patch File Download Diff File Expand all files Collapse all files

40
Backend/realtime/gateway.ts
View File

@@ -5,7 +5,7 @@ import { z } from 'zod';
import { db } from '../db/client';
import { simpleStreamingEnabled } from '../media/config';
import { deviceCommands, devices, streamSessions } from '../db/schema';
import { commands, devices, streamSessions } from '../db/schema';
import { canRelayWebrtcSignal } from '../streaming/simple';
import { hasRequiredTables } from '../utils/db-schema';
import { verifyDeviceToken } from '../utils/device-token';
@@ -97,8 +97,8 @@ const emitCommand = (command: {
};
export const dispatchCommandById = async (commandId: string): Promise<void> => {
const command = await db.query.deviceCommands.findFirst({
where: eq(deviceCommands.id, commandId),
const command = await db.query.commands.findFirst({
where: eq(commands.id, commandId),
});
if (!command) {
@@ -109,13 +109,13 @@ export const dispatchCommandById = async (commandId: string): Promise<void> => {
if (simpleStreamingEnabled && command.commandType === 'start_stream') {
await db
.update(deviceCommands)
.update(commands)
.set({
status: 'failed',
updatedAt: now,
error: 'start_stream command delivery disabled by SIMPLE_STREAMING',
})
.where(eq(deviceCommands.id, command.id));
.where(eq(commands.id, command.id));
return;
}
@@ -129,7 +129,7 @@ export const dispatchCommandById = async (commandId: string): Promise<void> => {
});
await db
.update(deviceCommands)
.update(commands)
.set({
status: delivered ? 'sent' : 'queued',
lastDispatchedAt: now,
@@ -137,14 +137,14 @@ export const dispatchCommandById = async (commandId: string): Promise<void> => {
updatedAt: now,
error: delivered ? null : 'target device offline',
})
.where(eq(deviceCommands.id, command.id));
.where(eq(commands.id, command.id));
};
const retryPendingCommands = async () => {
const threshold = new Date(Date.now() - RETRY_DISPATCH_DELAY_MS);
const pending = await db.query.deviceCommands.findMany({
where: and(eq(deviceCommands.status, 'sent'), lt(deviceCommands.lastDispatchedAt, threshold)),
const pending = await db.query.commands.findMany({
where: and(eq(commands.status, 'sent'), lt(commands.lastDispatchedAt, threshold)),
limit: 100,
orderBy: (fields, operators) => [operators.asc(fields.createdAt)],
});
@@ -154,13 +154,13 @@ const retryPendingCommands = async () => {
if (simpleStreamingEnabled && command.commandType === 'start_stream') {
await db
.update(deviceCommands)
.update(commands)
.set({
status: 'failed',
updatedAt: now,
error: 'start_stream retries disabled by SIMPLE_STREAMING',
})
.where(eq(deviceCommands.id, command.id));
.where(eq(commands.id, command.id));
continue;
}
@@ -168,13 +168,13 @@ const retryPendingCommands = async () => {
if (nextRetryCount > MAX_RETRIES) {
await db
.update(deviceCommands)
.update(commands)
.set({
status: 'failed',
updatedAt: now,
error: 'max retries exceeded',
})
.where(eq(deviceCommands.id, command.id));
.where(eq(commands.id, command.id));
continue;
}
@@ -188,7 +188,7 @@ const retryPendingCommands = async () => {
});
await db
.update(deviceCommands)
.update(commands)
.set({
status: delivered ? 'sent' : 'queued',
lastDispatchedAt: now,
@@ -196,7 +196,7 @@ const retryPendingCommands = async () => {
updatedAt: now,
error: delivered ? null : 'target device offline',
})
.where(eq(deviceCommands.id, command.id));
.where(eq(commands.id, command.id));
}
};
@@ -286,8 +286,8 @@ export const setupRealtimeGateway = (server: HttpServer): SocketIOServer => {
return;
}
const command = await db.query.deviceCommands.findFirst({
where: eq(deviceCommands.id, parsed.data.commandId),
const command = await db.query.commands.findFirst({
where: eq(commands.id, parsed.data.commandId),
});
if (!command) {
@@ -303,14 +303,14 @@ export const setupRealtimeGateway = (server: HttpServer): SocketIOServer => {
const now = new Date();
await db
.update(deviceCommands)
.update(commands)
.set({
status: parsed.data.status,
acknowledgedAt: now,
updatedAt: now,
error: parsed.data.status === 'rejected' ? parsed.data.error ?? 'rejected' : null,
})
.where(eq(deviceCommands.id, command.id));
.where(eq(commands.id, command.id));
io?.to(roomForDevice(command.sourceDeviceId)).emit('command:status', {
commandId: command.id,
@@ -371,7 +371,7 @@ export const setupRealtimeGateway = (server: HttpServer): SocketIOServer => {
});
if (!retryTimer) {
const requiredTables = ['device_commands'];
const requiredTables = ['commands'];
void (async () => {
const ready = await hasRequiredTables(requiredTables);