format issues + socket

This commit is contained in:
Daniel Bulant 2026-04-21 22:01:53 +02:00
parent 7738645a69
commit f79a9893a1
No known key found for this signature in database
16 changed files with 1984 additions and 1654 deletions

6
api/AGENTS.md Normal file
View file

@ -0,0 +1,6 @@
Run biome and typescript checks after your changes:
```
bun x biome ci
bun x tsc --noEmit
```

View file

@ -1,11 +1,17 @@
import { defineConfig } from "drizzle-kit";
const databaseUrl = process.env.DATABASE_URL;
if (!databaseUrl) {
throw new Error("Missing required env var: DATABASE_URL");
}
export default defineConfig({
out: "./drizzle",
schema: "./src/db/schema.ts",
dialect: "postgresql",
dbCredentials: {
url: process.env.DATABASE_URL!,
url: databaseUrl,
},
schemaFilter: ["public"],
});

View file

@ -5,7 +5,8 @@
"private": true,
"main": "src/index.ts",
"scripts": {
"dev": "bun run --watch src/index.ts"
"dev": "bun run --watch src/index.ts",
"typecheck": "tsc --noEmit"
},
"devDependencies": {
"@types/bun": "latest"

View file

@ -1,12 +1,18 @@
import { SpotifyApi } from "@spotify/web-api-ts-sdk";
import { betterAuth } from "better-auth";
import { drizzleAdapter } from "better-auth/adapters/drizzle";
import Elysia from "elysia";
import { db } from "./db";
import Elysia, { status, type Context } from "elysia";
import * as schema from "./db/auth-schema";
import { SpotifyApi } from "@spotify/web-api-ts-sdk";
export const SPOTIFY_CLIENT_ID = process.env.SPOTIFY_CLIENT_ID!;
export const SPOTIFY_CLIENT_SECRET = process.env.SPOTIFY_CLIENT_SECRET!;
const requireEnv = (name: string): string => {
const value = process.env[name];
if (!value) throw new Error(`Missing required env var: ${name}`);
return value;
};
export const SPOTIFY_CLIENT_ID = requireEnv("SPOTIFY_CLIENT_ID");
export const SPOTIFY_CLIENT_SECRET = requireEnv("SPOTIFY_CLIENT_SECRET");
export const defaultSdk = SpotifyApi.withClientCredentials(
SPOTIFY_CLIENT_ID,

View file

@ -1,5 +1,5 @@
import { relations } from "drizzle-orm/_relations";
import { pgTable, text, timestamp, boolean, index } from "drizzle-orm/pg-core";
import { boolean, index, pgTable, text, timestamp } from "drizzle-orm/pg-core";
export const user = pgTable("user", {
id: text("id").primaryKey(),

View file

@ -1,4 +1,10 @@
import { drizzle } from "drizzle-orm/node-postgres";
import { relations } from "./schema";
export const db = drizzle(process.env.DATABASE_URL!, { relations });
const databaseUrl = process.env.DATABASE_URL;
if (!databaseUrl) {
throw new Error("Missing required env var: DATABASE_URL");
}
export const db = drizzle(databaseUrl, { relations });

View file

@ -13,6 +13,7 @@ import {
uuid,
} from "drizzle-orm/pg-core";
import { user } from "./auth-schema";
export * from "./auth-schema";
export const partyStatus = pgEnum("party_status", [

View file

@ -9,6 +9,8 @@ import type {
SimplifiedArtist,
Track,
} from "@spotify/web-api-ts-sdk";
import { and, eq, inArray, sql } from "drizzle-orm";
import { defaultSdk } from "../auth";
import { db } from ".";
import {
album,
@ -20,8 +22,8 @@ import {
artistImage,
followedArtist,
genre,
playbackHistory,
platformImage,
playbackHistory,
savedAlbum,
savedTrack,
topArtist,
@ -29,19 +31,23 @@ import {
track,
trackArtist,
} from "./schema";
import { and, eq, inArray, sql } from "drizzle-orm";
import { defaultSdk } from "../auth";
export const PLATFORM_SPOTIFY = "spotify" as const;
type DbClient = typeof db;
type DbTransaction = Parameters<typeof db.transaction>[0] extends (
tx: infer T,
) => Promise<any>
) => Promise<unknown>
? T
: never;
type DbLike = DbClient | DbTransaction;
const requireMapEntry = <K, V>(map: Map<K, V>, key: K, label: string): V => {
const value = map.get(key);
if (!value) throw new Error(`Missing ${label} for ${String(key)}`);
return value;
};
export async function upsertImages(images: Image[], dbClient: DbLike = db) {
await dbClient
.insert(platformImage)
@ -66,7 +72,7 @@ export async function upsertArtists(artists: Artist[], dbClient: DbLike = db) {
await dbClient
.insert(artist)
.values(
artists.map(({ id, name, images, genres, popularity, type }) => ({
artists.map(({ id, name, popularity, type }) => ({
platform: PLATFORM_SPOTIFY,
platform_id: id,
name,
@ -135,7 +141,7 @@ async function lookupMissingArtists(
) {
const missingArtistIds = await getMissingArtists(artistIds, dbClient);
if (missingArtistIds.length === 0) return [];
let missingArtists: Artist[] = [];
const missingArtists: Artist[] = [];
for (let i = 0; i < missingArtistIds.length / 50; i++) {
missingArtists.push(
...(await defaultSdk.artists.get(
@ -150,7 +156,9 @@ async function lookupMissingArtists(
function isFullArtistArray(
artists: Artist[] | SimplifiedArtist[],
): artists is Artist[] {
return "images" in artists[0]!;
const firstArtist = artists[0];
if (!firstArtist) return false;
return "images" in firstArtist;
}
async function upsertMissingArtists(
@ -192,7 +200,11 @@ async function getAlbumIdMap(albumIds: string[], dbClient: DbLike = db) {
.select({ id: album.id, platform_id: album.platform_id })
.from(album)
.where(inArray(album.platform_id, albumIds));
return new Map(rows.map((row) => [row.platform_id ?? "", row.id]));
return new Map(
rows
.filter((row) => row.platform_id)
.map((row) => [row.platform_id as string, row.id]),
);
}
async function getTrackIdMap(trackIds: string[], dbClient: DbLike = db) {
@ -201,7 +213,11 @@ async function getTrackIdMap(trackIds: string[], dbClient: DbLike = db) {
.select({ id: track.id, platform_id: track.platform_id })
.from(track)
.where(inArray(track.platform_id, trackIds));
return new Map(rows.map((row) => [row.platform_id ?? "", row.id]));
return new Map(
rows
.filter((row) => row.platform_id)
.map((row) => [row.platform_id as string, row.id]),
);
}
export async function upsertAlbums(
@ -309,7 +325,7 @@ export async function upsertTracks(tracks: Track[], dbClient: DbLike = db) {
.insert(track)
.values(
tracks.map((spotifyTrack) => ({
albumId: albumIdMap.get(spotifyTrack.album.id)!,
albumId: requireMapEntry(albumIdMap, spotifyTrack.album.id, "albumId"),
name: spotifyTrack.name,
platform: PLATFORM_SPOTIFY,
platform_id: spotifyTrack.id,
@ -359,7 +375,7 @@ export async function upsertTopArtists(
.insert(topArtist)
.values(
artists.map((spotifyArtist, index) => ({
artistId: artistIdMap.get(spotifyArtist.id)!,
artistId: requireMapEntry(artistIdMap, spotifyArtist.id, "artistId"),
position: index + 1,
userId,
timeline,
@ -384,7 +400,7 @@ export async function upsertTopTracks(
.insert(topTrack)
.values(
tracks.map((spotifyTrack, index) => ({
trackId: trackIdMap.get(spotifyTrack.id)!,
trackId: requireMapEntry(trackIdMap, spotifyTrack.id, "trackId"),
position: index + 1,
userId,
timeline,
@ -409,7 +425,7 @@ export async function upsertSavedAlbums(
.insert(savedAlbum)
.values(
saved.map((item) => ({
albumId: albumIdMap.get(item.album.id)!,
albumId: requireMapEntry(albumIdMap, item.album.id, "albumId"),
userId,
saved_at: new Date(item.added_at),
})),
@ -433,7 +449,7 @@ export async function upsertSavedTracks(
.insert(savedTrack)
.values(
saved.map((item) => ({
trackId: trackIdMap.get(item.track.id)!,
trackId: requireMapEntry(trackIdMap, item.track.id, "trackId"),
userId,
saved_at: new Date(item.added_at),
})),
@ -456,7 +472,7 @@ export async function upsertFollowedArtists(
.insert(followedArtist)
.values(
artists.map((spotifyArtist) => ({
artistId: artistIdMap.get(spotifyArtist.id)!,
artistId: requireMapEntry(artistIdMap, spotifyArtist.id, "artistId"),
userId,
})),
)
@ -479,7 +495,7 @@ export async function upsertPlaybackHistory(
.insert(playbackHistory)
.values(
items.map((item) => ({
trackId: trackIdMap.get(item.track.id)!,
trackId: requireMapEntry(trackIdMap, item.track.id, "trackId"),
userId,
played_at: new Date(item.played_at),
})),

View file

@ -1,11 +1,11 @@
import { Elysia, t } from "elysia";
import { DBOS } from "@dbos-inc/dbos-sdk";
import { Elysia } from "elysia";
import { betterAuthElysia } from "./auth";
import { syncApp } from "./routes/sync";
import { DBOS } from "@dbos-inc/dbos-sdk";
import "./workflows/sync";
import "./dbos.ts";
import { statsApp } from "./routes/stats.ts";
import { partyApp } from "./routes/party";
import { statsApp } from "./routes/stats.ts";
const app = new Elysia()
.use(betterAuthElysia)

85
api/src/party-sockets.ts Normal file
View file

@ -0,0 +1,85 @@
type PartySocketEvent = {
type: string;
[key: string]: unknown;
};
type WebSocketLike = {
send: (data: string) => void;
close?: (code?: number, reason?: string) => void;
};
const partySockets = new Map<string, Map<string, Set<WebSocketLike>>>();
function getPartyUserSockets(partyId: string, userId: string) {
const partyMap = partySockets.get(partyId);
if (!partyMap) return null;
return partyMap.get(userId) ?? null;
}
export function registerPartySocket(
partyId: string,
userId: string,
ws: WebSocketLike,
) {
let partyMap = partySockets.get(partyId);
if (!partyMap) {
partyMap = new Map();
partySockets.set(partyId, partyMap);
}
let userSockets = partyMap.get(userId);
if (!userSockets) {
userSockets = new Set();
partyMap.set(userId, userSockets);
}
userSockets.add(ws);
}
export function unregisterPartySocket(
partyId: string,
userId: string,
ws: WebSocketLike,
) {
const partyMap = partySockets.get(partyId);
if (!partyMap) return;
const userSockets = partyMap.get(userId);
if (!userSockets) return;
userSockets.delete(ws);
if (userSockets.size === 0) {
partyMap.delete(userId);
}
if (partyMap.size === 0) {
partySockets.delete(partyId);
}
}
export function broadcastPartyEvent(partyId: string, event: PartySocketEvent) {
const partyMap = partySockets.get(partyId);
if (!partyMap) return;
const payload = JSON.stringify(event);
for (const userSockets of partyMap.values()) {
for (const ws of userSockets) {
ws.send(payload);
}
}
}
export function sendPartyEventToUser(
partyId: string,
userId: string,
event: PartySocketEvent,
) {
const userSockets = getPartyUserSockets(partyId, userId);
if (!userSockets) return;
const payload = JSON.stringify(event);
for (const ws of userSockets) {
ws.send(payload);
}
}

View file

@ -1,8 +1,14 @@
import Elysia, { t } from "elysia";
import { and, eq } from "drizzle-orm";
import { betterAuthElysia } from "../auth";
import Elysia, { t } from "elysia";
import { auth, betterAuthElysia } from "../auth";
import { db } from "../db";
import { party, partyMember } from "../db/schema";
import {
broadcastPartyEvent,
registerPartySocket,
sendPartyEventToUser,
unregisterPartySocket,
} from "../party-sockets";
const PARTY_STATUS = ["created", "started", "ended"] as const;
@ -11,11 +17,29 @@ type PartyStatus = (typeof PARTY_STATUS)[number];
type DbClient = typeof db;
type DbTransaction = Parameters<typeof db.transaction>[0] extends (
tx: infer T,
) => Promise<any>
) => Promise<unknown>
? T
: never;
type DbLike = DbClient | DbTransaction;
type PartySnapshot = NonNullable<Awaited<ReturnType<typeof getPartyStatus>>>;
type PartySocketMessage =
| {
type: "member_payload";
payload: unknown;
}
| {
type: "ping";
};
const MAX_MEMBER_PAYLOAD_SIZE = 8_000;
type PartyWsData = {
user?: { id: string };
partyId?: string;
};
async function getPartyForUser(userId: string) {
const memberships = await db.query.partyMember.findMany({
where: {
@ -63,6 +87,23 @@ async function getPartyStatus(partyId: string) {
};
}
function broadcastSnapshot(partyId: string, snapshot: PartySnapshot | null) {
if (!snapshot) return;
broadcastPartyEvent(partyId, {
type: "party_status",
party: snapshot.party,
members: snapshot.members,
});
}
function getPayloadSize(payload: unknown) {
try {
return JSON.stringify(payload).length;
} catch {
return Infinity;
}
}
async function cleanupPartyIfEmpty(dbClient: DbLike, partyId: string) {
const members = await dbClient.query.partyMember.findMany({
where: {
@ -86,6 +127,7 @@ async function leaveParty(dbClient: DbLike, userId: string) {
joinedAt: "asc",
},
});
let newHostId: string | null = null;
if (nextHost) {
const currentParty = await dbClient.query.party.findFirst({
where: {
@ -100,10 +142,14 @@ async function leaveParty(dbClient: DbLike, userId: string) {
lastUpdated: new Date(),
})
.where(eq(party.id, member.partyId));
newHostId = nextHost.userId;
}
}
await cleanupPartyIfEmpty(dbClient, member.partyId);
return member.partyId;
return {
partyId: member.partyId,
newHostId,
};
}
function isValidStatus(status: string): status is PartyStatus {
@ -114,6 +160,101 @@ export const partyApp = new Elysia()
.use(betterAuthElysia)
.group("/party", (app) =>
app
.ws("/ws", {
beforeHandle: async ({ request, set }) => {
const session = await auth.api.getSession({
headers: request.headers,
});
if (!session) {
set.status = 401;
return;
}
return {
user: session.user,
session: session.session,
};
},
open: async (ws) => {
const data = ws.data as unknown as PartyWsData;
const user = data.user;
if (!user) return;
const membership = await getMemberRecord(db, user.id);
if (!membership) {
ws.send(
JSON.stringify({
type: "error",
message: "You are not in a party.",
}),
);
ws.close?.(1008, "Not in a party");
return;
}
const snapshot = await getPartyStatus(membership.partyId);
data.partyId = membership.partyId;
registerPartySocket(membership.partyId, user.id, ws);
if (snapshot) {
ws.send(
JSON.stringify({
type: "snapshot",
party: snapshot.party,
members: snapshot.members,
}),
);
}
},
message: async (ws, message: PartySocketMessage) => {
const data = ws.data as unknown as PartyWsData;
const user = data.user;
if (!user) return;
if (message.type === "ping") {
ws.send(JSON.stringify({ type: "pong" }));
return;
}
if (message.type !== "member_payload") return;
const membership = await getMemberRecord(db, user.id);
if (!membership) return;
if (getPayloadSize(message.payload) > MAX_MEMBER_PAYLOAD_SIZE) {
ws.send(
JSON.stringify({
type: "error",
message: "Payload too large.",
}),
);
return;
}
const currentParty = await db.query.party.findFirst({
where: { id: membership.partyId },
});
if (!currentParty) return;
sendPartyEventToUser(membership.partyId, currentParty.hostId, {
type: "member_payload",
fromUserId: user.id,
payload: message.payload,
});
},
close: async (ws) => {
const data = ws.data as unknown as PartyWsData;
const user = data.user;
const { partyId } = data;
if (!user) return;
if (!partyId) {
const membership = await getMemberRecord(db, user.id);
if (!membership) return;
unregisterPartySocket(membership.partyId, user.id, ws);
return;
}
unregisterPartySocket(partyId, user.id, ws);
},
body: t.Union([
t.Object({ type: t.Literal("ping") }),
t.Object({ type: t.Literal("member_payload"), payload: t.Any() }),
]),
})
.get(
"/status",
async ({ user }) => {
@ -138,9 +279,11 @@ export const partyApp = new Elysia()
return { error: "Target user not found." };
}
const { partyId, hostChanged, leaveResult } = await db.transaction(
async (tx) => {
const leaveResult = await leaveParty(tx, user.id);
let partyId: string | null = null;
await db.transaction(async (tx) => {
await leaveParty(tx, user.id);
let hostChanged = false;
const targetMembership = await getMemberRecord(tx, targetUserId);
if (targetMembership) {
@ -152,6 +295,7 @@ export const partyApp = new Elysia()
lastUpdated: new Date(),
})
.where(eq(party.id, partyId));
hostChanged = true;
} else {
const created = await tx
.insert(party)
@ -160,21 +304,61 @@ export const partyApp = new Elysia()
hostId: targetUserId,
})
.returning({ id: party.id });
partyId = created[0]!.id;
const createdId = created[0]?.id ?? null;
if (!createdId) {
return {
partyId: null,
hostChanged,
leaveResult,
};
}
partyId = createdId;
await tx.insert(partyMember).values({
partyId,
userId: targetUserId,
});
}
if (!partyId) {
return {
partyId: null,
hostChanged,
leaveResult,
};
}
await tx
.insert(partyMember)
.values({ partyId, userId: user.id })
.onConflictDoNothing();
});
return {
partyId,
hostChanged,
leaveResult,
};
},
);
if (!partyId) return { party: null, members: [] };
const status = await getPartyStatus(partyId);
if (leaveResult?.newHostId) {
broadcastPartyEvent(leaveResult.partyId, {
type: "host_changed",
hostId: leaveResult.newHostId,
});
}
if (hostChanged) {
broadcastPartyEvent(partyId, {
type: "host_changed",
hostId: targetUserId,
});
}
broadcastPartyEvent(partyId, {
type: "member_joined",
userId: user.id,
});
broadcastSnapshot(partyId, status);
return status ?? { party: null, members: [] };
},
{
@ -187,11 +371,22 @@ export const partyApp = new Elysia()
.post(
"/leave",
async ({ user }) => {
const partyId = await db.transaction(async (tx) => {
const result = await db.transaction(async (tx) => {
return await leaveParty(tx, user.id);
});
if (!partyId) return { party: null, members: [] };
const status = await getPartyStatus(partyId);
if (!result) return { party: null, members: [] };
const status = await getPartyStatus(result.partyId);
broadcastPartyEvent(result.partyId, {
type: "member_left",
userId: user.id,
});
if (result.newHostId) {
broadcastPartyEvent(result.partyId, {
type: "host_changed",
hostId: result.newHostId,
});
}
broadcastSnapshot(result.partyId, status);
return status ?? { party: null, members: [] };
},
{ auth: true },
@ -232,6 +427,12 @@ export const partyApp = new Elysia()
await cleanupPartyIfEmpty(tx, currentMembership.partyId);
});
const status = await getPartyStatus(currentMembership.partyId);
broadcastPartyEvent(currentMembership.partyId, {
type: "member_left",
userId: body.memberUserId,
kickedBy: user.id,
});
broadcastSnapshot(currentMembership.partyId, status);
return status ?? { party: null, members: [] };
},
{
@ -285,6 +486,7 @@ export const partyApp = new Elysia()
});
const status = await getPartyStatus(currentMembership.partyId);
broadcastSnapshot(currentMembership.partyId, status);
return status ?? { party: null, members: [] };
},
{

View file

@ -1,5 +1,5 @@
import Elysia from "elysia";
import { sql } from "drizzle-orm";
import Elysia from "elysia";
import { betterAuthElysia } from "../auth";
import { db } from "../db";
import {

View file

@ -1,5 +1,4 @@
import { DBOS, ConfiguredInstance } from "@dbos-inc/dbos-sdk";
import { SpotifyApi } from "@spotify/web-api-ts-sdk";
import { ConfiguredInstance, DBOS } from "@dbos-inc/dbos-sdk";
import type {
Artist,
PlayHistory,
@ -7,6 +6,8 @@ import type {
SavedTrack,
Track,
} from "@spotify/web-api-ts-sdk";
import { SpotifyApi } from "@spotify/web-api-ts-sdk";
import { eq } from "drizzle-orm";
import { auth, SPOTIFY_CLIENT_ID } from "../auth";
import { db } from "../db";
import {
@ -24,7 +25,6 @@ import {
upsertTopArtists,
upsertTopTracks,
} from "../db/spotify";
import { eq } from "drizzle-orm";
const timelines = ["short_term", "medium_term", "long_term"] as const;
type Timeline = (typeof timelines)[number];
@ -39,10 +39,6 @@ type SyncPayload = {
};
export class SpotifySyncWorkflow extends ConfiguredInstance {
constructor(name: string) {
super(name);
}
@DBOS.workflow()
async syncUser(userId: string) {
console.log("Sync start");
@ -186,7 +182,8 @@ export class SpotifySyncWorkflow extends ConfiguredInstance {
const artists = page.artists;
followed.push(...artists.items);
if (!artists.next || artists.items.length === 0) break;
after = artists.items[artists.items.length - 1]!.id;
const lastArtist = artists.items.at(-1);
after = lastArtist?.id;
}
return followed;
}
@ -235,8 +232,12 @@ export class SpotifySyncWorkflow extends ConfiguredInstance {
});
return SpotifyApi.withAccessToken(SPOTIFY_CLIENT_ID, {
access_token: accessToken.accessToken,
expires_in: Date.now() - Number(accessToken.accessTokenExpiresAt!),
expires: Number(accessToken.accessTokenExpiresAt),
expires_in: accessToken.accessTokenExpiresAt
? Date.now() - Number(accessToken.accessTokenExpiresAt)
: 0,
expires: accessToken.accessTokenExpiresAt
? Number(accessToken.accessTokenExpiresAt)
: 0,
refresh_token: "",
token_type: "",
});

View file

@ -25,6 +25,6 @@
// Some stricter flags (disabled by default)
"noUnusedLocals": false,
"noUnusedParameters": false,
"noPropertyAccessFromIndexSignature": false,
},
"noPropertyAccessFromIndexSignature": false
}
}