300 lines
8.3 KiB
JavaScript
300 lines
8.3 KiB
JavaScript
import express from 'express';
|
|
import multer from 'multer';
|
|
import { Queue } from 'bullmq';
|
|
import { v4 as uuidv4 } from 'uuid';
|
|
import pool from '../db/pool.js';
|
|
import { s3Client, uploadStream, deleteObject, getS3Bucket } from '../s3/client.js';
|
|
import {
|
|
CreateMultipartUploadCommand,
|
|
UploadPartCommand,
|
|
CompleteMultipartUploadCommand,
|
|
AbortMultipartUploadCommand,
|
|
} from '@aws-sdk/client-s3';
|
|
import { getAmppConfig, ensureFolderPath } from '../ampp/client.js';
|
|
|
|
const router = express.Router();
|
|
|
|
const memoryStorage = multer.memoryStorage();
|
|
const upload = multer({ storage: memoryStorage });
|
|
|
|
const parseRedisUrl = (url) => {
|
|
const parsed = new URL(url);
|
|
return { host: parsed.hostname, port: parseInt(parsed.port, 10) };
|
|
};
|
|
|
|
// Only proxy queue needed here — proxy worker dispatches thumbnail once done
|
|
const proxyQueue = new Queue('proxy', {
|
|
connection: parseRedisUrl(process.env.REDIS_URL || 'redis://queue:6379'),
|
|
});
|
|
|
|
// ---------------------------------------------------------------
|
|
// AMPP Sync Helpers
|
|
// ---------------------------------------------------------------
|
|
|
|
async function resolveBinPath(binId) {
|
|
const segments = [];
|
|
let currentId = binId;
|
|
while (currentId) {
|
|
const result = await pool.query(
|
|
'SELECT id, name, parent_id FROM bins WHERE id = $1',
|
|
[currentId]
|
|
);
|
|
if (result.rows.length === 0) break;
|
|
const bin = result.rows[0];
|
|
segments.unshift(bin.name);
|
|
currentId = bin.parent_id;
|
|
}
|
|
return segments;
|
|
}
|
|
|
|
/**
|
|
* Fire-and-forget: mirror asset's project/bin path into AMPP folder hierarchy.
|
|
* Never throws — failures are logged but never surface to the caller.
|
|
*/
|
|
async function syncToAmpp(assetId, projectId, binId) {
|
|
try {
|
|
const config = await getAmppConfig();
|
|
if (!config) return;
|
|
|
|
const projResult = await pool.query(
|
|
'SELECT name FROM projects WHERE id = $1',
|
|
[projectId]
|
|
);
|
|
if (projResult.rows.length === 0) return;
|
|
const projectName = projResult.rows[0].name;
|
|
|
|
const segments = [projectName];
|
|
if (binId) {
|
|
const binSegments = await resolveBinPath(binId);
|
|
segments.push(...binSegments);
|
|
}
|
|
|
|
const folderId = await ensureFolderPath(config, segments);
|
|
if (!folderId) return;
|
|
|
|
await pool.query(
|
|
'UPDATE assets SET ampp_folder_id = $1, ampp_synced_at = NOW() WHERE id = $2',
|
|
[folderId, assetId]
|
|
);
|
|
|
|
console.log(`[AMPP] asset ${assetId} → folder ${folderId} (${segments.join(' / ')})`);
|
|
} catch (err) {
|
|
console.error(`[AMPP] sync failed for asset ${assetId}:`, err.message);
|
|
}
|
|
}
|
|
|
|
// Derive a media_type string from a MIME type
|
|
function mediaTypeFromMime(mime = '') {
|
|
if (mime.startsWith('video')) return 'video';
|
|
if (mime.startsWith('audio')) return 'audio';
|
|
if (mime.startsWith('image')) return 'image';
|
|
return 'document';
|
|
}
|
|
|
|
// POST /api/v1/upload/init - Initialize a multipart upload
|
|
router.post('/init', async (req, res, next) => {
|
|
try {
|
|
const { filename, fileSize, contentType, projectId, binId, tags } = req.body;
|
|
|
|
if (!filename || !fileSize || !contentType || !projectId) {
|
|
return res.status(400).json({
|
|
error: 'Missing required fields: filename, fileSize, contentType, projectId',
|
|
});
|
|
}
|
|
|
|
const assetId = uuidv4();
|
|
const s3Key = `originals/${assetId}/${filename}`;
|
|
const tagsArray = tags ? (Array.isArray(tags) ? tags : [tags]) : [];
|
|
|
|
await pool.query(
|
|
`INSERT INTO assets (
|
|
id, project_id, bin_id, filename, display_name, status,
|
|
media_type, original_s3_key, file_size, tags, created_at, updated_at
|
|
)
|
|
VALUES ($1,$2,$3,$4,$4,'ingesting',$5,$6,$7,$8,NOW(),NOW())`,
|
|
[
|
|
assetId, projectId, binId || null, filename,
|
|
mediaTypeFromMime(contentType),
|
|
s3Key, fileSize,
|
|
tagsArray.length > 0 ? tagsArray : null,
|
|
]
|
|
);
|
|
|
|
const multipartUpload = await s3Client.send(
|
|
new CreateMultipartUploadCommand({
|
|
Bucket: getS3Bucket(),
|
|
Key: s3Key,
|
|
ContentType: contentType,
|
|
})
|
|
);
|
|
|
|
res.json({
|
|
assetId,
|
|
uploadId: multipartUpload.UploadId,
|
|
key: s3Key,
|
|
});
|
|
} catch (err) {
|
|
next(err);
|
|
}
|
|
});
|
|
|
|
// POST /api/v1/upload/part - Upload a single part
|
|
router.post('/part', upload.single('file'), async (req, res, next) => {
|
|
try {
|
|
const { uploadId, key, partNumber } = req.body;
|
|
|
|
if (!uploadId || !key || !partNumber || !req.file) {
|
|
return res.status(400).json({
|
|
error: 'Missing required fields: uploadId, key, partNumber, and file',
|
|
});
|
|
}
|
|
|
|
const partUpload = await s3Client.send(
|
|
new UploadPartCommand({
|
|
Bucket: getS3Bucket(),
|
|
Key: key,
|
|
PartNumber: parseInt(partNumber, 10),
|
|
UploadId: uploadId,
|
|
Body: req.file.buffer,
|
|
})
|
|
);
|
|
|
|
res.json({
|
|
partNumber: parseInt(partNumber, 10),
|
|
etag: partUpload.ETag,
|
|
});
|
|
} catch (err) {
|
|
next(err);
|
|
}
|
|
});
|
|
|
|
// POST /api/v1/upload/complete - Complete the multipart upload
|
|
router.post('/complete', async (req, res, next) => {
|
|
try {
|
|
const { uploadId, key, assetId, parts } = req.body;
|
|
|
|
if (!uploadId || !key || !assetId || !parts || !Array.isArray(parts)) {
|
|
return res.status(400).json({
|
|
error: 'Missing required fields: uploadId, key, assetId, and parts array',
|
|
});
|
|
}
|
|
|
|
await s3Client.send(
|
|
new CompleteMultipartUploadCommand({
|
|
Bucket: getS3Bucket(),
|
|
Key: key,
|
|
UploadId: uploadId,
|
|
MultipartUpload: {
|
|
Parts: parts.map(p => ({
|
|
ETag: p.ETag || p.etag,
|
|
PartNumber: p.partNumber || p.PartNumber,
|
|
})),
|
|
},
|
|
})
|
|
);
|
|
|
|
const result = await pool.query(
|
|
`UPDATE assets
|
|
SET status = 'processing', updated_at = NOW()
|
|
WHERE id = $1 RETURNING *`,
|
|
[assetId]
|
|
);
|
|
|
|
if (result.rows.length === 0) {
|
|
return res.status(404).json({ error: 'Asset not found' });
|
|
}
|
|
|
|
const asset = result.rows[0];
|
|
|
|
await proxyQueue.add('generate', {
|
|
assetId,
|
|
inputKey: key,
|
|
outputKey: `proxies/${assetId}.mp4`,
|
|
});
|
|
|
|
syncToAmpp(asset.id, asset.project_id, asset.bin_id);
|
|
|
|
res.json(asset);
|
|
} catch (err) {
|
|
next(err);
|
|
}
|
|
});
|
|
|
|
// POST /api/v1/upload/abort - Abort upload
|
|
router.post('/abort', async (req, res, next) => {
|
|
try {
|
|
const { uploadId, key, assetId } = req.body;
|
|
|
|
if (!uploadId || !key || !assetId) {
|
|
return res.status(400).json({
|
|
error: 'Missing required fields: uploadId, key, assetId',
|
|
});
|
|
}
|
|
|
|
await s3Client.send(
|
|
new AbortMultipartUploadCommand({ Bucket: getS3Bucket(), Key: key, UploadId: uploadId })
|
|
);
|
|
|
|
await pool.query('DELETE FROM assets WHERE id = $1', [assetId]);
|
|
|
|
res.json({ message: 'Upload aborted and asset deleted' });
|
|
} catch (err) {
|
|
next(err);
|
|
}
|
|
});
|
|
|
|
// POST /api/v1/upload/simple - Single-file upload for smaller files (<50 MB)
|
|
router.post('/simple', upload.single('file'), async (req, res, next) => {
|
|
try {
|
|
const { filename, projectId, binId, tags, contentType } = req.body;
|
|
|
|
if (!filename || !projectId || !req.file) {
|
|
return res.status(400).json({
|
|
error: 'Missing required fields: filename, projectId, and file',
|
|
});
|
|
}
|
|
|
|
const assetId = uuidv4();
|
|
const s3Key = `originals/${assetId}/${filename}`;
|
|
const mimeType = contentType || req.file.mimetype;
|
|
const tagsArray = tags ? (Array.isArray(tags) ? tags : [tags]) : [];
|
|
|
|
await pool.query(
|
|
`INSERT INTO assets (
|
|
id, project_id, bin_id, filename, display_name, status,
|
|
media_type, original_s3_key, file_size, tags, created_at, updated_at
|
|
)
|
|
VALUES ($1,$2,$3,$4,$4,'ingesting',$5,$6,$7,$8,NOW(),NOW())`,
|
|
[
|
|
assetId, projectId, binId || null, filename,
|
|
mediaTypeFromMime(mimeType),
|
|
s3Key, req.file.size,
|
|
tagsArray.length > 0 ? tagsArray : null,
|
|
]
|
|
);
|
|
|
|
await uploadStream(s3Key, req.file.buffer, mimeType);
|
|
|
|
const result = await pool.query(
|
|
`UPDATE assets SET status = 'processing', updated_at = NOW()
|
|
WHERE id = $1 RETURNING *`,
|
|
[assetId]
|
|
);
|
|
|
|
const asset = result.rows[0];
|
|
|
|
await proxyQueue.add('generate', {
|
|
assetId,
|
|
inputKey: s3Key,
|
|
outputKey: `proxies/${assetId}.mp4`,
|
|
});
|
|
|
|
syncToAmpp(assetId, projectId, binId || null);
|
|
|
|
res.json(asset);
|
|
} catch (err) {
|
|
next(err);
|
|
}
|
|
});
|
|
|
|
export default router;
|