simeonradivoev/gameflow-deck
simeonradivoev/gameflow-deck
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat: implemented a basic store and emulatorjs

Browse source
This commit is contained in:
Simeon Radivoev 2026-03-14 02:15:57 +02:00
parent 2f32cbc730
commit 7286541822
Signed by: simeonradivoev
GPG key ID: 7611A451D2A5D37A
121 changed files with 5900 additions and 1092 deletions
Download patch file Download diff file Expand all files Collapse all files

254
src/bun/api/jobs/install-job.ts
View file

@ -2,13 +2,16 @@ import { IJob, JobContext } from "../task-queue";
import { mkdir } from 'node:fs/promises';
import { and, eq, or } from 'drizzle-orm';
import fs from 'node:fs/promises';
import * as schema from "../schema/app";
import * as emulatorSchema from "../schema/emulators";
import * as schema from "@schema/app";
import * as emulatorSchema from "@schema/emulators";
import path from 'node:path';
import { getPlatformApiPlatformsIdGet, getRomApiRomsIdGet } from "@clients/romm";
import { getPlatformApiPlatformsIdGet, getRomApiRomsIdGet, PlatformSchema } from "@clients/romm";
import { config, db, emulatorsDb, jar } from "../app";
import unzip from 'unzip-stream';
import { Readable, Transform } from "node:stream";
import { extractStoreGameSourceId, getStoreGameFromId } from "../store/services/gamesService";
import * as igdb from 'ts-igdb-client';
import secrets from "../secrets";
interface JobConfig
{
@ -18,15 +21,15 @@ interface JobConfig
export class InstallJob implements IJob
{
public id: number;
public gameId: string;
public source: string;
public sourceId: number;
public sourceId: string;
public config?: JobConfig;
static id = "install-job" as const;
constructor(id: number, source: string, sourceId: number, config?: JobConfig)
constructor(id: string, source: string, sourceId: string, config?: JobConfig)
{
this.id = id;
this.gameId = id;
this.config = config;
this.sourceId = sourceId;
this.source = source;
@ -41,6 +44,65 @@ export class InstallJob implements IJob
{
const downloadPath = config.get('downloadPath');
let downloadUrl: URL;
let cookie: string = '';
let screenshotUrls: string[];
let coverUrl: string;
let rommPlatform: PlatformSchema | undefined;
let slug: string | null;
let path_fs: string | undefined;
let summary: string | null;
let name: string | null;
let last_played: Date | null;
let igdb_id: number | null;
let ra_id: number | null;
let source_id: string;
let system_slug: string;
let extract_path: string;
switch (this.source)
{
case 'romm':
const rom = (await getRomApiRomsIdGet({ path: { id: Number(this.gameId) }, throwOnError: true })).data;
rommPlatform = (await getPlatformApiPlatformsIdGet({ path: { id: rom.platform_id }, throwOnError: true })).data;
const rommAddress = config.get('rommAddress');
coverUrl = `${rommAddress}${rom.path_cover_large}`;
screenshotUrls = rom.merged_screenshots.map(s => `${config.get('rommAddress')}${s}`);
last_played = rom.rom_user.last_played ? new Date(rom.rom_user.last_played) : null;
igdb_id = rom.igdb_id;
ra_id = rom.ra_id;
summary = rom.summary;
name = rom.name;
path_fs = path.join(rom.fs_path, rom.fs_name);
source_id = String(rom.id);
slug = rom.slug;
system_slug = rommPlatform.slug;
extract_path = '';
downloadUrl = new URL(`${config.get('rommAddress')}/api/roms/download`);
downloadUrl.searchParams.set('rom_ids', String(this.gameId));
cookie = await jar.getCookieString(config.get('rommAddress') ?? '');
break;
case 'store':
const game = await getStoreGameFromId(this.gameId);
const gameId = extractStoreGameSourceId(this.gameId);
coverUrl = game.pictures.titlescreens[0];
screenshotUrls = game.pictures.screenshots;
downloadUrl = new URL(game.file);
slug = this.gameId;
source_id = this.gameId;
name = game.title;
summary = game.description;
system_slug = gameId.system;
extract_path = 'roms', gameId.system;
break;
default:
throw new Error("Unsupported source");
}
if (this.config?.dryDownload !== true)
{
/*
@ -92,11 +154,10 @@ export class InstallJob implements IJob
await fs.rm(zipFilePath);*/
cx.setProgress(0, 'download');
const downloadUrl = new URL(`${config.get('rommAddress')}/api/roms/download`);
downloadUrl.searchParams.set('rom_ids', String(this.id));
const res = await fetch(downloadUrl, {
headers: {
cookie: await jar.getCookieString(config.get('rommAddress') ?? '')
cookie: cookie
},
});
@ -119,62 +180,99 @@ export class InstallJob implements IJob
await new Promise((resolve, reject) =>
{
Readable.fromWeb(res.body as any).pipe(progressStream).pipe(unzip.Extract({ path: downloadPath })).on('close', resolve).on('error', reject);
const extract = unzip.Extract({ path: path.join(downloadPath, extract_path), });
(extract as any).unzipStream.on('entry', (entry: any) =>
{
if (!path_fs)
path_fs = path.join(extract_path, entry.path);
});
Readable.fromWeb(res.body as any).pipe(progressStream)
.pipe(extract)
.on('close', resolve)
.on('error', reject);
});
}
const rom = (await getRomApiRomsIdGet({ path: { id: this.id }, throwOnError: true })).data;
const romPlatform = (await getPlatformApiPlatformsIdGet({ path: { id: rom.platform_id }, throwOnError: true })).data;
if (this.config?.dryDownload === true)
{
rom.files.length;
await mkdir(path.join(downloadPath, rom.fs_path, rom.fs_name), { recursive: true });
await mkdir(path.join(downloadPath, extract_path), { recursive: true });
}
// pre-fetch screenshots
const screenshots = await Promise.all(rom.merged_screenshots.map(s => fetch(`${config.get('rommAddress')}${s}`)));
const rommAddress = config.get('rommAddress');
const coverResponse = await fetch(`${rommAddress}${rom.path_cover_large}`);
const coverResponse = await fetch(coverUrl);
const cover = Buffer.from(await coverResponse.arrayBuffer());
if (cx.abortSignal.aborted) return;
await db.transaction(async (tx) =>
{
// Search for existing platform
const platformSearch = [];
if (romPlatform.igdb_id) platformSearch.push(eq(schema.platforms.igdb_id, romPlatform.igdb_id));
if (romPlatform.igdb_slug) platformSearch.push(eq(schema.platforms.igdb_slug, romPlatform.igdb_slug));
if (romPlatform.ra_id) platformSearch.push(eq(schema.platforms.ra_id, romPlatform.ra_id));
if (romPlatform.slug) platformSearch.push(eq(schema.platforms.slug, romPlatform.slug));
if (romPlatform.moby_id) platformSearch.push(eq(schema.platforms.moby_id, romPlatform.moby_id));
const platformSearch = [eq(schema.platforms.slug, system_slug)];
const esPlatformSearch = [eq(emulatorSchema.systemMappings.system, system_slug)];
const esPlatform = await emulatorsDb
.select({ slug: emulatorSchema.systemMappings.system, romm_slug: emulatorSchema.systemMappings.sourceSlug })
.from(emulatorSchema.systemMappings)
.where(and(eq(emulatorSchema.systemMappings.source, 'romm'), eq(emulatorSchema.systemMappings.sourceSlug, romPlatform.slug)));
if (rommPlatform)
{
if (rommPlatform.igdb_id) platformSearch.push(eq(schema.platforms.igdb_id, rommPlatform.igdb_id));
if (rommPlatform.igdb_slug) platformSearch.push(eq(schema.platforms.igdb_slug, rommPlatform.igdb_slug));
if (rommPlatform.ra_id) platformSearch.push(eq(schema.platforms.ra_id, rommPlatform.ra_id));
if (rommPlatform.moby_id) platformSearch.push(eq(schema.platforms.moby_id, rommPlatform.moby_id));
const existingPlatform = await tx.query.platforms.findFirst({ where: or(...platformSearch) });
esPlatformSearch.push(eq(emulatorSchema.systemMappings.source, 'romm'));
esPlatformSearch.push(eq(emulatorSchema.systemMappings.sourceSlug, rommPlatform.slug));
}
const esPlatform = await emulatorsDb.query.systemMappings.findFirst({
with: { system: true },
where: and(...esPlatformSearch)
});
if (esPlatform)
platformSearch.push(eq(schema.platforms.es_slug, esPlatform.system.name));
let existingPlatform = await tx.query.platforms.findFirst({ where: or(...platformSearch) });
let platformId: number;
if (!existingPlatform)
{
// Create new local platform
const platformCover = await fetch(`${rommAddress}/assets/platforms/${romPlatform.slug.toLocaleLowerCase()}.svg`);
const platform: typeof schema.platforms.$inferInsert = {
slug: romPlatform.slug,
igdb_id: romPlatform.igdb_id,
igdb_slug: romPlatform.igdb_slug,
ra_id: romPlatform.ra_id,
cover: Buffer.from(await platformCover.arrayBuffer()),
cover_type: platformCover.headers.get('content-type'),
name: romPlatform.name,
family_name: romPlatform.family_name,
es_slug: esPlatform.length > 0 ? esPlatform[0].slug : undefined
};
// TODO: add ES slug once I have better way to query ES
const [{ id }] = await tx.insert(schema.platforms).values(platform).returning({ id: schema.platforms.id });
platformId = id;
// TODO: use something else than the romm demo as CDN
const platformCover = await fetch(`https://demo.romm.app/assets/platforms/${system_slug}.svg`);
if (!esPlatform && !rommPlatform)
{
// go to unknown platform
existingPlatform = await tx.query.platforms.findFirst({ where: eq(schema.platforms.slug, "unknown") });
if (existingPlatform)
{
platformId = existingPlatform.id;
} else
{
const [{ id }] = await tx.insert(schema.platforms).values({
slug: 'unknown',
name: "Unknown"
}).returning({ id: schema.platforms.id });
platformId = id;
}
} else
{
// Create new local platform
const platform: typeof schema.platforms.$inferInsert = {
slug: rommPlatform?.slug ?? esPlatform?.system.name ?? '',
igdb_id: rommPlatform?.igdb_id,
igdb_slug: rommPlatform?.igdb_slug,
ra_id: rommPlatform?.ra_id,
cover: Buffer.from(await platformCover.arrayBuffer()),
cover_type: platformCover.headers.get('content-type'),
name: rommPlatform?.name ?? esPlatform?.system.fullname ?? '',
family_name: rommPlatform?.family_name,
es_slug: esPlatform?.system.name ?? undefined
};
// TODO: add ES slug once I have better way to query ES
const [{ id }] = await tx.insert(schema.platforms).values(platform).returning({ id: schema.platforms.id });
platformId = id;
}
} else
{
platformId = existingPlatform.id;
@ -182,32 +280,52 @@ export class InstallJob implements IJob
// create the rom
const game: typeof schema.games.$inferInsert = {
source_id: rom.id,
source: 'romm',
slug: rom.slug,
path_fs: path.join(rom.fs_path, rom.fs_name),
last_played: rom.rom_user.last_played ? new Date(rom.rom_user.last_played) : null,
source_id,
source: this.source,
slug,
path_fs,
last_played: last_played,
platform_id: platformId,
igdb_id: rom.igdb_id,
ra_id: rom.ra_id,
summary: rom.summary,
name: rom.name,
cover: Buffer.from(await coverResponse.arrayBuffer()),
igdb_id: igdb_id,
ra_id: ra_id,
summary: summary,
name,
cover,
cover_type: coverResponse.headers.get('content-type')
};
// Save screenshots and update database
const [{ id }] = await tx.insert(schema.games).values(game).returning({ id: schema.games.id });
await tx.insert(schema.screenshots).values(await Promise.all(screenshots.map(async (response) =>
{
const screenshot: typeof schema.screenshots.$inferInsert = {
game_id: id,
content: Buffer.from(await response.arrayBuffer()),
type: response.headers.get('content-type')
};
return screenshot;
})));
if (screenshotUrls.length <= 0 && process.env.TWITCH_CLIENT_ID)
{
const access_token = await secrets.get({ service: 'gamflow_twitch', name: 'access_token' });
if (access_token)
{
const client = igdb.igdb(process.env.TWITCH_CLIENT_ID, access_token);
const { data } = await client.request('artworks').pipe(igdb.fields(['game', 'url']), igdb.where('game', '=', igdb_id)).execute();
screenshotUrls.push(...data.filter(s => s.url).map(s => s.url!));
}
}
// pre-fetch screenshots
const screenshots = await Promise.all(screenshotUrls.map(s => fetch(s)));
if (screenshots.length > 0)
{
await tx.insert(schema.screenshots).values(await Promise.all(screenshots.map(async (response) =>
{
const screenshot: typeof schema.screenshots.$inferInsert = {
game_id: id,
content: Buffer.from(await response.arrayBuffer()),
type: response.headers.get('content-type')
};
return screenshot;
})));
}
});
}

80
src/bun/api/jobs/jobs.ts Normal file
View file

@ -0,0 +1,80 @@
import Elysia from "elysia";
import z, { } from "zod";
import { taskQueue } from "../app";
import { LoginJob } from "./login-job";
import TwitchLoginJob from "./twitch-login-job";
import UpdateStoreJob from "./update-store";
function registerJob<const Path extends string, TS, T extends { id: Path, dataSchema?: TS; }> (job: T, path: Path, dataSchema: TS)
{
return new Elysia().ws(path, {
body: z.discriminatedUnion('type', [
z.object({ type: z.literal('cancel') })
]),
response: z.discriminatedUnion('type', [
z.object({
type: z.literal(['data', 'started', 'progress']),
status: z.string(),
progress: z.number(),
data: dataSchema
}),
z.object({ type: z.literal(['completed', 'ended']) }),
z.object({ type: z.literal('error'), error: z.unknown() })
]),
open (ws)
{
const job = taskQueue.findJob(path);
if (job)
{
ws.send({ type: 'data', status: job.status, progress: job.progress, data: job.job.exposeData?.() });
}
(ws.data as any).cleanup = [
taskQueue.on('started', ({ id, job }) =>
{
if (id === path)
{
ws.send({ type: 'started', status: job.status, progress: job.progress, data: job.job.exposeData?.() });
}
}),
taskQueue.on('progress', ({ id, job }) =>
{
if (id === path)
{
ws.send({ type: 'started', status: job.status, progress: job.progress, data: job.job.exposeData?.() });
}
}),
taskQueue.on('completed', ({ id }) =>
{
if (id === path)
{
ws.send({ type: 'completed' });
}
}),
taskQueue.on('error', ({ id, error }) =>
{
if (id === path)
{
ws.send({ type: 'error', error: error });
}
})
];
},
close (ws)
{
(ws.data as any).cleanup.forEach((d: Function) => d());
},
message (ws, message)
{
if (message.type === 'cancel')
{
taskQueue.findJob(path)?.abort('cancel');
}
},
});
}
export const jobs = new Elysia({ prefix: '/api/jobs' })
.use(registerJob(LoginJob, LoginJob.id, LoginJob.dataSchema))
.use(registerJob(TwitchLoginJob, TwitchLoginJob.id, TwitchLoginJob.dataSchema))
.use(registerJob(UpdateStoreJob, UpdateStoreJob.id, undefined));

18
src/bun/api/jobs/login-job.ts
View file

@ -4,20 +4,27 @@ import { LOGIN_PORT, SERVER_URL } from "@/shared/constants";
import { host, localIp } from "@/bun/utils/host";
import cors from "@elysiajs/cors";
import { tryLoginAndSave } from "../auth";
import z from "zod";
import { config } from "../app";
import z from "zod";
import { delay } from "@/shared/utils";
export class LoginJob implements IJob
{
endsAt: Date;
startedAt: Date;
url: string;
static id = "login-job" as const;
static dataSchema = z.object({ endsAt: z.date(), startedAt: z.date(), url: z.url() });
constructor()
{
this.endsAt = new Date();
this.endsAt = new Date(new Date().getTime() + 300000);
this.startedAt = new Date();
this.url = `http://${localIp}:${LOGIN_PORT}/`;
}
exposeData = (): z.infer<typeof LoginJob.dataSchema> => ({ endsAt: this.endsAt, startedAt: this.startedAt, url: this.url });
async start (context: JobContext): Promise<any>
{
const loginServer = new Elysia({ serve: { hostname: localIp, port: LOGIN_PORT } })
@ -44,12 +51,7 @@ export class LoginJob implements IJob
try
{
loginServer.listen({});
await new Promise((resolve, reject) =>
{
this.endsAt = new Date(new Date().getTime() + 300000);
context.abortSignal.addEventListener('abort', () => reject());
setTimeout(() => { reject('timeout'); }, 300000); // auto close after 5 minutes
});
await delay(this.endsAt, context.abortSignal);
} catch
{
} finally

110
src/bun/api/jobs/twitch-login-job.ts Normal file
View file

@ -0,0 +1,110 @@
import { IJob, JobContext } from "../task-queue";
import secrets from "../secrets";
import open from "open";
import z from "zod";
import { delay } from "@/shared/utils";
interface TwitchDevice
{
device_code: string,
expires_in: number,
expires_at: Date,
started_at: Date,
interval: number,
user_code: string,
verification_uri: string;
}
export default class TwitchLoginJob implements IJob
{
twitchScopes = "analytics:read:extensions analytics:read:games user:read:email";
device?: TwitchDevice;
clientId: string;
openInBrowser: boolean;
static id = 'twitch-login-job' as const;
static dataSchema = z.object({ expires_at: z.date(), started_at: z.date(), url: z.url(), user_code: z.string() }).or(z.undefined());
constructor(clientId: string, openInBrowser: boolean)
{
this.clientId = clientId;
this.openInBrowser = openInBrowser;
}
exposeData = (): z.infer<typeof TwitchLoginJob.dataSchema> => this.device ? ({
expires_at: this.device.expires_at,
started_at: this.device.started_at,
url: this.device.verification_uri,
user_code: this.device.user_code
}) : undefined;
async start (context: JobContext): Promise<any>
{
context.setProgress(0, "Retrieving Device");
let res = await fetch("https://id.twitch.tv/oauth2/device", {
method: "POST",
body: new URLSearchParams({
client_id: this.clientId,
scopes: this.twitchScopes
}),
signal: context.abortSignal
});
const device: TwitchDevice = await res.json();
const expiredTimeout = setTimeout(() => context.abort('expired'), device.expires_in * 1000);
device.expires_at = new Date(new Date().getTime() + device.expires_in * 1000);
device.started_at = new Date();
this.device = device;
try
{
if (this.openInBrowser)
open(device.verification_uri);
this.device = device;
context.setProgress(50, "Waiting For Authentication");
while (true)
{
if (context.abortSignal.aborted) break;
await delay(device.interval * 1000, context.abortSignal);
res = await fetch("https://id.twitch.tv/oauth2/token", {
method: "POST",
body: new URLSearchParams({
client_id: this.clientId,
scopes: this.twitchScopes,
device_code: this.device.device_code,
grant_type: "urn:ietf:params:oauth:grant-type:device_code"
}),
signal: context.abortSignal
});
if (res.status === 200)
{
const data: {
access_token: string,
expires_in: number,
refresh_token: string,
scope: string[],
token_type: string;
} = await res.json();
secrets.set({ service: 'gamflow_twitch', name: 'access_token', value: data.access_token });
secrets.set({ service: 'gamflow_twitch', name: 'refresh_token', value: data.refresh_token });
secrets.set({ service: 'gamflow_twitch', name: 'expires_in', value: new Date(new Date().getTime() + data.expires_in).toString() });
break;
}
else if (res.status !== 400)
{
console.error(res.statusText);
break;
}
}
} finally
{
clearTimeout(expiredTimeout);
}
}
}

76
src/bun/api/jobs/update-store.ts Normal file
View file

@ -0,0 +1,76 @@
import { ensureDir } from "fs-extra";
import { IJob, JobContext } from "../task-queue";
import { getStoreFolder } from "../store/store";
export default class UpdateStoreJob implements IJob
{
static id = "update-store" as const;
static origin = "https://github.com/simeonradivoev/gameflow-store.git";
static branch = "master";
async gitCommand (commands: string[], dir: string)
{
const proc = Bun.spawn(['git', ...commands], {
cwd: dir,
stdout: "pipe",
stderr: "pipe",
});
const [output] = await Promise.all([
new Response(proc.stdout).text(),
proc.exited,
]);
return output.trim();
}
async isGitRepo (dir: string)
{
return (await this.gitCommand(["rev-parse", "--is-inside-work-tree"], dir)) === 'true';
}
async getOrigin (dir: string)
{
const origin = await this.gitCommand(["remote", "get-url", "origin"], dir);
return origin;
}
async hasChanges (dir: string)
{
return (await this.gitCommand(["status", "--porcelain"], dir)).length > 0;
}
async start (context: JobContext)
{
const storeFolder = getStoreFolder();
await ensureDir(storeFolder);
context.setProgress(10);
if (await this.isGitRepo(storeFolder))
{
const existingOrigin = await this.getOrigin(storeFolder);
if (existingOrigin !== UpdateStoreJob.origin)
{
throw new Error(`Git Repo in downloads is not valid. It has origin of ${existingOrigin}. Repo must be of ${UpdateStoreJob.origin}`);
}
// check for uncommitted changes
const status = await this.gitCommand([" status", "--porcelain"], storeFolder);
if (status.length > 0)
{
console.log("Cleaning local changes...");
await this.gitCommand(["reset", "--hard"], storeFolder);
await this.gitCommand(["clean", "-fd"], storeFolder);
}
// fetch & reset to remote
await this.gitCommand(["fetch", "origin"], storeFolder);
await this.gitCommand(["reset", "--hard", `origin/${UpdateStoreJob.branch}`], storeFolder);
console.log("Shop Repo updated");
} else
{
context.setProgress(50);
await this.gitCommand(["clone", "--depth", "1", "--branch", UpdateStoreJob.branch, UpdateStoreJob.origin, '.'], storeFolder);
context.setProgress(100);
}
}
}