Phases 2-4: Auth, providers, stream management
This commit is contained in:
267
src/routes/streams/lifecycle.ts
Normal file
267
src/routes/streams/lifecycle.ts
Normal file
@@ -0,0 +1,267 @@
|
||||
import { FastifyPluginAsync } from 'fastify';
|
||||
import { requireAuth } from '../../middleware/require-auth.js';
|
||||
import { decrypt, encrypt } from '../../services/crypto.service.js';
|
||||
import {
|
||||
createYouTubeBroadcast,
|
||||
transitionYouTubeBroadcast,
|
||||
refreshYouTubeToken,
|
||||
} from '../../services/youtube.service.js';
|
||||
import {
|
||||
getTwitchStreamKey,
|
||||
updateTwitchChannel,
|
||||
refreshTwitchToken,
|
||||
} from '../../services/twitch.service.js';
|
||||
import { AppError } from '../../plugins/error-handler.js';
|
||||
import type { PrepareResponse, PreparedDestination } from '../../types/api.js';
|
||||
|
||||
const TWITCH_RTMP_URL = 'rtmp://live.twitch.tv/app';
|
||||
|
||||
async function getDecryptedToken(
|
||||
prisma: any,
|
||||
userId: string,
|
||||
serviceId: string,
|
||||
): Promise<{ account: any; accessToken: string }> {
|
||||
const account = await prisma.linkedAccount.findUnique({
|
||||
where: { userId_serviceId: { userId, serviceId } },
|
||||
});
|
||||
if (!account) throw new AppError(400, `No ${serviceId} account linked`);
|
||||
|
||||
// Lazy refresh if token is expired or about to expire
|
||||
if (account.tokenExpiresAt < new Date(Date.now() + 60 * 1000)) {
|
||||
const refreshToken = decrypt(account.refreshTokenEnc, account.refreshTokenIv);
|
||||
let newAccess: string;
|
||||
let newRefresh: string | undefined;
|
||||
let expiresIn: number;
|
||||
|
||||
if (serviceId === 'YOUTUBE') {
|
||||
const result = await refreshYouTubeToken(refreshToken);
|
||||
newAccess = result.accessToken;
|
||||
expiresIn = result.expiresIn;
|
||||
} else {
|
||||
const result = await refreshTwitchToken(refreshToken);
|
||||
newAccess = result.accessToken;
|
||||
newRefresh = result.refreshToken;
|
||||
expiresIn = result.expiresIn;
|
||||
}
|
||||
|
||||
const accessEnc = encrypt(newAccess);
|
||||
const updateData: any = {
|
||||
accessTokenEnc: accessEnc.ciphertext,
|
||||
accessTokenIv: accessEnc.iv,
|
||||
tokenExpiresAt: new Date(Date.now() + expiresIn * 1000),
|
||||
};
|
||||
if (newRefresh) {
|
||||
const refreshEnc = encrypt(newRefresh);
|
||||
updateData.refreshTokenEnc = refreshEnc.ciphertext;
|
||||
updateData.refreshTokenIv = refreshEnc.iv;
|
||||
}
|
||||
|
||||
await prisma.linkedAccount.update({
|
||||
where: { id: account.id },
|
||||
data: updateData,
|
||||
});
|
||||
|
||||
return { account, accessToken: newAccess };
|
||||
}
|
||||
|
||||
return {
|
||||
account,
|
||||
accessToken: decrypt(account.accessTokenEnc, account.accessTokenIv),
|
||||
};
|
||||
}
|
||||
|
||||
const lifecycleRoutes: FastifyPluginAsync = async (fastify) => {
|
||||
// POST /streams/plans/:id/prepare — create broadcasts, get RTMP info
|
||||
fastify.post<{ Params: { id: string } }>('/streams/plans/:id/prepare', {
|
||||
preHandler: [requireAuth],
|
||||
schema: {
|
||||
params: {
|
||||
type: 'object',
|
||||
required: ['id'],
|
||||
properties: { id: { type: 'string' } },
|
||||
},
|
||||
},
|
||||
}, async (request) => {
|
||||
const plan = await fastify.prisma.streamPlan.findFirst({
|
||||
where: { id: request.params.id, userId: request.userId },
|
||||
include: { destinations: true },
|
||||
});
|
||||
if (!plan) throw new AppError(404, 'Stream plan not found');
|
||||
if (plan.status !== 'DRAFT') {
|
||||
throw new AppError(400, `Plan is already ${plan.status}`);
|
||||
}
|
||||
|
||||
const prepared: PreparedDestination[] = [];
|
||||
|
||||
for (const dest of plan.destinations) {
|
||||
const { account, accessToken } = await getDecryptedToken(
|
||||
fastify.prisma,
|
||||
request.userId,
|
||||
dest.serviceId,
|
||||
);
|
||||
|
||||
if (dest.serviceId === 'YOUTUBE') {
|
||||
const broadcast = await createYouTubeBroadcast(
|
||||
accessToken,
|
||||
dest.title,
|
||||
dest.description,
|
||||
dest.privacyStatus,
|
||||
);
|
||||
|
||||
await fastify.prisma.streamDestination.update({
|
||||
where: { id: dest.id },
|
||||
data: {
|
||||
rtmpUrl: broadcast.rtmpUrl,
|
||||
streamKey: broadcast.streamKey,
|
||||
broadcastId: broadcast.id,
|
||||
status: 'READY',
|
||||
},
|
||||
});
|
||||
|
||||
prepared.push({
|
||||
serviceId: 'YOUTUBE',
|
||||
rtmpUrl: broadcast.rtmpUrl,
|
||||
streamKey: broadcast.streamKey,
|
||||
broadcastId: broadcast.id,
|
||||
});
|
||||
} else if (dest.serviceId === 'TWITCH') {
|
||||
// Update channel info
|
||||
const tags = dest.tags ? dest.tags.split(',').map((t) => t.trim()).filter(Boolean) : [];
|
||||
await updateTwitchChannel(
|
||||
accessToken,
|
||||
account.accountId,
|
||||
dest.title,
|
||||
dest.gameId,
|
||||
tags,
|
||||
);
|
||||
|
||||
// Get stream key
|
||||
const streamKey = await getTwitchStreamKey(accessToken, account.accountId);
|
||||
|
||||
await fastify.prisma.streamDestination.update({
|
||||
where: { id: dest.id },
|
||||
data: {
|
||||
rtmpUrl: TWITCH_RTMP_URL,
|
||||
streamKey,
|
||||
broadcastId: account.accountId,
|
||||
status: 'READY',
|
||||
},
|
||||
});
|
||||
|
||||
prepared.push({
|
||||
serviceId: 'TWITCH',
|
||||
rtmpUrl: TWITCH_RTMP_URL,
|
||||
streamKey,
|
||||
broadcastId: account.accountId,
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
await fastify.prisma.streamPlan.update({
|
||||
where: { id: plan.id },
|
||||
data: { status: 'READY' },
|
||||
});
|
||||
|
||||
const response: PrepareResponse = {
|
||||
planId: plan.id,
|
||||
destinations: prepared,
|
||||
};
|
||||
return response;
|
||||
});
|
||||
|
||||
// POST /streams/plans/:id/start — transition to LIVE
|
||||
fastify.post<{ Params: { id: string } }>('/streams/plans/:id/start', {
|
||||
preHandler: [requireAuth],
|
||||
schema: {
|
||||
params: {
|
||||
type: 'object',
|
||||
required: ['id'],
|
||||
properties: { id: { type: 'string' } },
|
||||
},
|
||||
},
|
||||
}, async (request) => {
|
||||
const plan = await fastify.prisma.streamPlan.findFirst({
|
||||
where: { id: request.params.id, userId: request.userId },
|
||||
include: { destinations: true },
|
||||
});
|
||||
if (!plan) throw new AppError(404, 'Stream plan not found');
|
||||
if (plan.status !== 'READY') {
|
||||
throw new AppError(400, `Plan must be READY to start, currently ${plan.status}`);
|
||||
}
|
||||
|
||||
for (const dest of plan.destinations) {
|
||||
if (dest.serviceId === 'YOUTUBE' && dest.broadcastId) {
|
||||
const { accessToken } = await getDecryptedToken(
|
||||
fastify.prisma,
|
||||
request.userId,
|
||||
'YOUTUBE',
|
||||
);
|
||||
await transitionYouTubeBroadcast(accessToken, dest.broadcastId, 'live');
|
||||
}
|
||||
// Twitch goes live automatically when RTMP stream is received
|
||||
|
||||
await fastify.prisma.streamDestination.update({
|
||||
where: { id: dest.id },
|
||||
data: { status: 'LIVE' },
|
||||
});
|
||||
}
|
||||
|
||||
await fastify.prisma.streamPlan.update({
|
||||
where: { id: plan.id },
|
||||
data: { status: 'LIVE' },
|
||||
});
|
||||
|
||||
return { success: true, status: 'LIVE' };
|
||||
});
|
||||
|
||||
// POST /streams/plans/:id/end — end stream
|
||||
fastify.post<{ Params: { id: string } }>('/streams/plans/:id/end', {
|
||||
preHandler: [requireAuth],
|
||||
schema: {
|
||||
params: {
|
||||
type: 'object',
|
||||
required: ['id'],
|
||||
properties: { id: { type: 'string' } },
|
||||
},
|
||||
},
|
||||
}, async (request) => {
|
||||
const plan = await fastify.prisma.streamPlan.findFirst({
|
||||
where: { id: request.params.id, userId: request.userId },
|
||||
include: { destinations: true },
|
||||
});
|
||||
if (!plan) throw new AppError(404, 'Stream plan not found');
|
||||
if (plan.status !== 'LIVE') {
|
||||
throw new AppError(400, `Plan must be LIVE to end, currently ${plan.status}`);
|
||||
}
|
||||
|
||||
for (const dest of plan.destinations) {
|
||||
if (dest.serviceId === 'YOUTUBE' && dest.broadcastId) {
|
||||
const { accessToken } = await getDecryptedToken(
|
||||
fastify.prisma,
|
||||
request.userId,
|
||||
'YOUTUBE',
|
||||
);
|
||||
try {
|
||||
await transitionYouTubeBroadcast(accessToken, dest.broadcastId, 'complete');
|
||||
} catch {
|
||||
// Non-fatal — stream may already be ended
|
||||
}
|
||||
}
|
||||
// Twitch ends when RTMP stream stops
|
||||
|
||||
await fastify.prisma.streamDestination.update({
|
||||
where: { id: dest.id },
|
||||
data: { status: 'ENDED' },
|
||||
});
|
||||
}
|
||||
|
||||
await fastify.prisma.streamPlan.update({
|
||||
where: { id: plan.id },
|
||||
data: { status: 'ENDED' },
|
||||
});
|
||||
|
||||
return { success: true, status: 'ENDED' };
|
||||
});
|
||||
};
|
||||
|
||||
export default lifecycleRoutes;
|
||||
Reference in New Issue
Block a user