Compare commits
2 commits
21910eca41
...
58878752a8
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
58878752a8 | ||
|
|
9072ce76ba |
11 changed files with 412 additions and 356 deletions
|
|
@ -52,7 +52,10 @@ export const partyMember = pgTable(
|
||||||
joinedAt: timestamp().defaultNow().notNull(),
|
joinedAt: timestamp().defaultNow().notNull(),
|
||||||
lastSeen: timestamp().defaultNow().notNull(),
|
lastSeen: timestamp().defaultNow().notNull(),
|
||||||
},
|
},
|
||||||
(partyMember) => [uniqueIndex().on(partyMember.partyId, partyMember.userId)],
|
(partyMember) => [
|
||||||
|
uniqueIndex().on(partyMember.partyId, partyMember.userId),
|
||||||
|
index().on(partyMember.userId, partyMember.joinedAt),
|
||||||
|
],
|
||||||
);
|
);
|
||||||
|
|
||||||
export const platform = pgEnum("enum_platform", ["spotify", "apple"]);
|
export const platform = pgEnum("enum_platform", ["spotify", "apple"]);
|
||||||
|
|
|
||||||
|
|
@ -7,7 +7,7 @@ import "./workflows/sync";
|
||||||
import "./workflows/party-analysis";
|
import "./workflows/party-analysis";
|
||||||
import "./dbos.ts";
|
import "./dbos.ts";
|
||||||
import { partyApp } from "./routes/party";
|
import { partyApp } from "./routes/party";
|
||||||
import { partySocketApp } from "./routes/party-socket";
|
import { partySocketApp, pubsub } from "./routes/party-socket";
|
||||||
import { statsApp } from "./routes/stats.ts";
|
import { statsApp } from "./routes/stats.ts";
|
||||||
|
|
||||||
const app = new Elysia()
|
const app = new Elysia()
|
||||||
|
|
@ -23,6 +23,8 @@ const app = new Elysia()
|
||||||
)
|
)
|
||||||
.listen(4000);
|
.listen(4000);
|
||||||
|
|
||||||
|
pubsub.setServer(app.server);
|
||||||
|
|
||||||
export type App = typeof app;
|
export type App = typeof app;
|
||||||
|
|
||||||
await DBOS.launch({
|
await DBOS.launch({
|
||||||
|
|
|
||||||
|
|
@ -1,6 +1,7 @@
|
||||||
import { eq } from "drizzle-orm";
|
import { eq } from "drizzle-orm";
|
||||||
import { db } from "./db";
|
import { db } from "./db";
|
||||||
import { party, partyMember } from "./db/schema";
|
import { party, partyMember } from "./db/schema";
|
||||||
|
import type { PartySnapshot } from "./party-types";
|
||||||
|
|
||||||
type DbClient = typeof db;
|
type DbClient = typeof db;
|
||||||
type DbTransaction = Parameters<typeof db.transaction>[0] extends (
|
type DbTransaction = Parameters<typeof db.transaction>[0] extends (
|
||||||
|
|
@ -29,11 +30,16 @@ export async function getMemberRecord(dbClient: DbLike, userId: string) {
|
||||||
where: {
|
where: {
|
||||||
userId,
|
userId,
|
||||||
},
|
},
|
||||||
|
orderBy: {
|
||||||
|
joinedAt: "desc",
|
||||||
|
},
|
||||||
})) ?? null
|
})) ?? null
|
||||||
);
|
);
|
||||||
}
|
}
|
||||||
|
|
||||||
export async function getPartyStatus(partyId: string) {
|
export async function getPartyStatus(
|
||||||
|
partyId: string,
|
||||||
|
): Promise<PartySnapshot | null> {
|
||||||
const partyRecord = await db.query.party.findFirst({
|
const partyRecord = await db.query.party.findFirst({
|
||||||
where: {
|
where: {
|
||||||
id: partyId,
|
id: partyId,
|
||||||
|
|
|
||||||
|
|
@ -1,159 +0,0 @@
|
||||||
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>>>();
|
|
||||||
const userSockets = new 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 registerUserSocket(userId: string, ws: WebSocketLike) {
|
|
||||||
let sockets = userSockets.get(userId);
|
|
||||||
if (!sockets) {
|
|
||||||
sockets = new Set();
|
|
||||||
userSockets.set(userId, sockets);
|
|
||||||
}
|
|
||||||
|
|
||||||
sockets.add(ws);
|
|
||||||
}
|
|
||||||
|
|
||||||
export function unregisterUserSocket(userId: string, ws: WebSocketLike) {
|
|
||||||
const sockets = userSockets.get(userId);
|
|
||||||
if (!sockets) return;
|
|
||||||
|
|
||||||
sockets.delete(ws);
|
|
||||||
|
|
||||||
if (sockets.size === 0) {
|
|
||||||
userSockets.delete(userId);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
export function unregisterUserSocketFromAllParties(
|
|
||||||
userId: string,
|
|
||||||
ws: WebSocketLike,
|
|
||||||
) {
|
|
||||||
for (const [partyId, partyMap] of partySockets) {
|
|
||||||
const userSockets = partyMap.get(userId);
|
|
||||||
if (!userSockets) continue;
|
|
||||||
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);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
export function sendDirectEventToUser(userId: string, event: PartySocketEvent) {
|
|
||||||
const sockets = userSockets.get(userId);
|
|
||||||
if (!sockets) return;
|
|
||||||
|
|
||||||
const payload = JSON.stringify(event);
|
|
||||||
for (const ws of sockets) {
|
|
||||||
ws.send(payload);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
export function reassignUserSocketsToParty(
|
|
||||||
userId: string,
|
|
||||||
partyId: string | null,
|
|
||||||
) {
|
|
||||||
for (const [existingPartyId, partyMap] of partySockets) {
|
|
||||||
if (!partyMap.has(userId)) continue;
|
|
||||||
partyMap.delete(userId);
|
|
||||||
if (partyMap.size === 0) {
|
|
||||||
partySockets.delete(existingPartyId);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
if (!partyId) return;
|
|
||||||
const sockets = userSockets.get(userId);
|
|
||||||
if (!sockets) return;
|
|
||||||
|
|
||||||
let partyMap = partySockets.get(partyId);
|
|
||||||
if (!partyMap) {
|
|
||||||
partyMap = new Map();
|
|
||||||
partySockets.set(partyId, partyMap);
|
|
||||||
}
|
|
||||||
|
|
||||||
partyMap.set(userId, new Set(sockets));
|
|
||||||
}
|
|
||||||
32
api/src/party-types.ts
Normal file
32
api/src/party-types.ts
Normal file
|
|
@ -0,0 +1,32 @@
|
||||||
|
import type { InferSelectModel } from "drizzle-orm";
|
||||||
|
import type { party, partyMember, user } from "./db/schema";
|
||||||
|
|
||||||
|
export type Party = InferSelectModel<typeof party>;
|
||||||
|
export type PartyMember = InferSelectModel<typeof partyMember>;
|
||||||
|
export type User = InferSelectModel<typeof user>;
|
||||||
|
|
||||||
|
export type PartyMemberWithUser = PartyMember & { user: User | null };
|
||||||
|
|
||||||
|
export const PARTY_STATUS = ["created", "started", "ended"] as const;
|
||||||
|
export type PartyStatus = (typeof PARTY_STATUS)[number];
|
||||||
|
|
||||||
|
export type PartySnapshot = {
|
||||||
|
party: Party;
|
||||||
|
members: PartyMemberWithUser[];
|
||||||
|
};
|
||||||
|
|
||||||
|
export type PartyState = {
|
||||||
|
party: Party | null;
|
||||||
|
members: PartyMemberWithUser[];
|
||||||
|
};
|
||||||
|
|
||||||
|
export type PartySocketOutgoing =
|
||||||
|
| { type: "ping" }
|
||||||
|
| { type: "member_payload"; payload: unknown };
|
||||||
|
|
||||||
|
export type PartySocketEvent =
|
||||||
|
| { type: "snapshot"; party: Party | null; members: PartyMemberWithUser[] }
|
||||||
|
| { type: "party_status"; party: Party; members: PartyMemberWithUser[] }
|
||||||
|
| { type: "member_payload"; fromUserId: string; payload: unknown }
|
||||||
|
| { type: "error"; message: string }
|
||||||
|
| { type: "pong" };
|
||||||
|
|
@ -1,62 +1,49 @@
|
||||||
import Elysia, { t } from "elysia";
|
import { Elysia } from "elysia";
|
||||||
import { auth, betterAuthElysia } from "../auth";
|
|
||||||
|
import { betterAuthElysia } from "../auth";
|
||||||
|
|
||||||
import { db } from "../db";
|
import { db } from "../db";
|
||||||
import { getMemberRecord, getPartyStatus } from "../party-data";
|
import { getMemberRecord, getPartyStatus } from "../party-data";
|
||||||
import {
|
|
||||||
registerPartySocket,
|
|
||||||
registerUserSocket,
|
|
||||||
sendPartyEventToUser,
|
|
||||||
unregisterPartySocket,
|
|
||||||
unregisterUserSocket,
|
|
||||||
unregisterUserSocketFromAllParties,
|
|
||||||
} from "../party-sockets";
|
|
||||||
|
|
||||||
type PartySocketMessage =
|
function userTopic(userId: string) {
|
||||||
| {
|
return `user:${userId}`;
|
||||||
type: "member_payload";
|
|
||||||
payload: unknown;
|
|
||||||
}
|
}
|
||||||
| {
|
|
||||||
type: "ping";
|
function partyTopic(partyId: string) {
|
||||||
|
return `party:${partyId}`;
|
||||||
|
}
|
||||||
|
|
||||||
|
const socketPartyId = new WeakMap<object, string>();
|
||||||
|
|
||||||
|
export const pubsub = {
|
||||||
|
_server: null as ReturnType<typeof Bun.serve> | null,
|
||||||
|
setServer(server: ReturnType<typeof Bun.serve> | null) {
|
||||||
|
this._server = server;
|
||||||
|
},
|
||||||
|
publish(topic: string, data: string) {
|
||||||
|
this._server?.publish(topic, data);
|
||||||
|
},
|
||||||
};
|
};
|
||||||
|
|
||||||
const MAX_MEMBER_PAYLOAD_SIZE = 8_000;
|
export const topic = {
|
||||||
|
user: userTopic,
|
||||||
type PartyWsData = {
|
party: partyTopic,
|
||||||
user?: { id: string };
|
|
||||||
partyId?: string | null;
|
|
||||||
};
|
};
|
||||||
|
|
||||||
function getPayloadSize(payload: unknown) {
|
|
||||||
try {
|
|
||||||
return JSON.stringify(payload).length;
|
|
||||||
} catch {
|
|
||||||
return Infinity;
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
export const partySocketApp = new Elysia()
|
export const partySocketApp = new Elysia()
|
||||||
.use(betterAuthElysia)
|
.use(betterAuthElysia)
|
||||||
.group("/party-socket", (app) =>
|
.group("/party-socket", (app) =>
|
||||||
app.ws("/ws", {
|
app
|
||||||
beforeHandle: async ({ request, set }) => {
|
.get("/test", () => ({ ok: 1 }))
|
||||||
const session = await auth.api.getSession({
|
.ws("/ws", {
|
||||||
headers: request.headers,
|
auth: true,
|
||||||
});
|
publishToSelf: true,
|
||||||
if (!session) {
|
|
||||||
set.status = 401;
|
|
||||||
return;
|
|
||||||
}
|
|
||||||
return {
|
|
||||||
user: session.user,
|
|
||||||
session: session.session,
|
|
||||||
};
|
|
||||||
},
|
|
||||||
open: async (ws) => {
|
open: async (ws) => {
|
||||||
const data = ws.data as unknown as PartyWsData;
|
const user = ws.data.user;
|
||||||
const user = data.user;
|
|
||||||
if (!user) return;
|
if (!user) return;
|
||||||
registerUserSocket(user.id, ws);
|
|
||||||
|
ws.subscribe(userTopic(user.id));
|
||||||
|
|
||||||
const membership = await getMemberRecord(db, user.id);
|
const membership = await getMemberRecord(db, user.id);
|
||||||
if (!membership) {
|
if (!membership) {
|
||||||
ws.send(
|
ws.send(
|
||||||
|
|
@ -69,9 +56,10 @@ export const partySocketApp = new Elysia()
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
socketPartyId.set(ws, membership.partyId);
|
||||||
|
ws.subscribe(partyTopic(membership.partyId));
|
||||||
|
|
||||||
const snapshot = await getPartyStatus(membership.partyId);
|
const snapshot = await getPartyStatus(membership.partyId);
|
||||||
data.partyId = membership.partyId;
|
|
||||||
registerPartySocket(membership.partyId, user.id, ws);
|
|
||||||
if (snapshot) {
|
if (snapshot) {
|
||||||
ws.send(
|
ws.send(
|
||||||
JSON.stringify({
|
JSON.stringify({
|
||||||
|
|
@ -82,56 +70,64 @@ export const partySocketApp = new Elysia()
|
||||||
);
|
);
|
||||||
}
|
}
|
||||||
},
|
},
|
||||||
message: async (ws, message: PartySocketMessage) => {
|
message: async (ws, message) => {
|
||||||
const data = ws.data as unknown as PartyWsData;
|
const data = ws.data;
|
||||||
const user = data.user;
|
const user = data.user;
|
||||||
if (!user) return;
|
if (!user) return;
|
||||||
if (message.type === "ping") {
|
|
||||||
|
if (typeof message !== "string") return;
|
||||||
|
|
||||||
|
let parsed: { type: string; payload?: unknown };
|
||||||
|
try {
|
||||||
|
parsed = JSON.parse(message);
|
||||||
|
} catch {
|
||||||
|
ws.send(JSON.stringify({ type: "error", message: "Invalid JSON" }));
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
if (parsed.type === "ping") {
|
||||||
ws.send(JSON.stringify({ type: "pong" }));
|
ws.send(JSON.stringify({ type: "pong" }));
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
|
||||||
if (message.type !== "member_payload") return;
|
if (parsed.type !== "member_payload") return;
|
||||||
const membership = await getMemberRecord(db, user.id);
|
|
||||||
if (!membership) return;
|
|
||||||
|
|
||||||
if (getPayloadSize(message.payload) > MAX_MEMBER_PAYLOAD_SIZE) {
|
const MAX_MEMBER_PAYLOAD_SIZE = 8_000;
|
||||||
|
const payloadString = JSON.stringify(parsed.payload);
|
||||||
|
if (payloadString.length > MAX_MEMBER_PAYLOAD_SIZE) {
|
||||||
ws.send(
|
ws.send(
|
||||||
JSON.stringify({
|
JSON.stringify({ type: "error", message: "Payload too large." }),
|
||||||
type: "error",
|
|
||||||
message: "Payload too large.",
|
|
||||||
}),
|
|
||||||
);
|
);
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
const membership = await getMemberRecord(db, user.id);
|
||||||
|
if (!membership) return;
|
||||||
|
|
||||||
const currentParty = await db.query.party.findFirst({
|
const currentParty = await db.query.party.findFirst({
|
||||||
where: { id: membership.partyId },
|
where: { id: membership.partyId },
|
||||||
});
|
});
|
||||||
if (!currentParty) return;
|
if (!currentParty) return;
|
||||||
|
|
||||||
sendPartyEventToUser(membership.partyId, currentParty.hostId, {
|
ws.publish(
|
||||||
|
partyTopic(membership.partyId),
|
||||||
|
JSON.stringify({
|
||||||
type: "member_payload",
|
type: "member_payload",
|
||||||
fromUserId: user.id,
|
fromUserId: user.id,
|
||||||
payload: message.payload,
|
payload: parsed.payload,
|
||||||
});
|
}),
|
||||||
},
|
);
|
||||||
close: async (ws) => {
|
},
|
||||||
const data = ws.data as unknown as PartyWsData;
|
close: async (ws) => {
|
||||||
const user = data.user;
|
const user = ws.data.user;
|
||||||
const { partyId } = data;
|
if (!user) return;
|
||||||
if (!user) return;
|
|
||||||
if (!partyId) {
|
ws.unsubscribe(userTopic(user.id));
|
||||||
unregisterUserSocketFromAllParties(user.id, ws);
|
|
||||||
unregisterUserSocket(user.id, ws);
|
const partyId = socketPartyId.get(ws);
|
||||||
return;
|
if (!partyId) return;
|
||||||
}
|
|
||||||
unregisterPartySocket(partyId, user.id, ws);
|
ws.unsubscribe(partyTopic(partyId));
|
||||||
unregisterUserSocket(user.id, ws);
|
},
|
||||||
},
|
|
||||||
body: t.Union([
|
|
||||||
t.Object({ type: t.Literal("ping") }),
|
|
||||||
t.Object({ type: t.Literal("member_payload"), payload: t.Any() }),
|
|
||||||
]),
|
|
||||||
}),
|
}),
|
||||||
);
|
);
|
||||||
|
|
|
||||||
|
|
@ -10,29 +10,29 @@ import {
|
||||||
getPartyStatus,
|
getPartyStatus,
|
||||||
leaveParty,
|
leaveParty,
|
||||||
} from "../party-data";
|
} from "../party-data";
|
||||||
import {
|
import type { PartySnapshot } from "../party-types";
|
||||||
broadcastPartyEvent,
|
import { pubsub, topic } from "./party-socket";
|
||||||
reassignUserSocketsToParty,
|
|
||||||
sendDirectEventToUser,
|
|
||||||
} from "../party-sockets";
|
|
||||||
|
|
||||||
const PARTY_STATUS = ["created", "started", "ended"] as const;
|
|
||||||
|
|
||||||
type PartyStatus = (typeof PARTY_STATUS)[number];
|
|
||||||
|
|
||||||
type PartySnapshot = NonNullable<Awaited<ReturnType<typeof getPartyStatus>>>;
|
|
||||||
|
|
||||||
function broadcastSnapshot(partyId: string, snapshot: PartySnapshot | null) {
|
function broadcastSnapshot(partyId: string, snapshot: PartySnapshot | null) {
|
||||||
if (!snapshot) return;
|
if (!snapshot) return;
|
||||||
broadcastPartyEvent(partyId, {
|
pubsub.publish(
|
||||||
|
topic.party(partyId),
|
||||||
|
JSON.stringify({
|
||||||
type: "party_status",
|
type: "party_status",
|
||||||
party: snapshot.party,
|
party: snapshot.party,
|
||||||
members: snapshot.members,
|
members: snapshot.members,
|
||||||
});
|
}),
|
||||||
|
);
|
||||||
}
|
}
|
||||||
|
|
||||||
function isValidStatus(status: string): status is PartyStatus {
|
function broadcastToUser(userId: string, event: Record<string, unknown>) {
|
||||||
return PARTY_STATUS.includes(status as PartyStatus);
|
pubsub.publish(topic.user(userId), JSON.stringify(event));
|
||||||
|
}
|
||||||
|
|
||||||
|
function isValidStatus(
|
||||||
|
status: string,
|
||||||
|
): status is import("../party-types").PartyStatus {
|
||||||
|
return ["created", "started", "ended"].includes(status);
|
||||||
}
|
}
|
||||||
|
|
||||||
export const partyApp = new Elysia()
|
export const partyApp = new Elysia()
|
||||||
|
|
@ -127,31 +127,19 @@ export const partyApp = new Elysia()
|
||||||
if (!partyId) return { party: null, members: [] };
|
if (!partyId) return { party: null, members: [] };
|
||||||
const status = await getPartyStatus(partyId);
|
const status = await getPartyStatus(partyId);
|
||||||
if (leaveResult?.newHostId) {
|
if (leaveResult?.newHostId) {
|
||||||
broadcastPartyEvent(leaveResult.partyId, {
|
broadcastSnapshot(leaveResult.partyId, status);
|
||||||
type: "host_changed",
|
|
||||||
hostId: leaveResult.newHostId,
|
|
||||||
});
|
|
||||||
}
|
}
|
||||||
if (hostChanged) {
|
if (hostChanged) {
|
||||||
broadcastPartyEvent(partyId, {
|
|
||||||
type: "host_changed",
|
|
||||||
hostId: targetUserId,
|
|
||||||
});
|
|
||||||
}
|
|
||||||
broadcastPartyEvent(partyId, {
|
|
||||||
type: "member_joined",
|
|
||||||
userId: user.id,
|
|
||||||
});
|
|
||||||
broadcastSnapshot(partyId, status);
|
broadcastSnapshot(partyId, status);
|
||||||
reassignUserSocketsToParty(user.id, partyId);
|
}
|
||||||
reassignUserSocketsToParty(targetUserId, partyId);
|
broadcastSnapshot(partyId, status);
|
||||||
if (status) {
|
if (status) {
|
||||||
sendDirectEventToUser(targetUserId, {
|
broadcastToUser(targetUserId, {
|
||||||
type: "party_status",
|
type: "party_status",
|
||||||
party: status.party,
|
party: status.party,
|
||||||
members: status.members,
|
members: status.members,
|
||||||
});
|
});
|
||||||
sendDirectEventToUser(user.id, {
|
broadcastToUser(user.id, {
|
||||||
type: "party_status",
|
type: "party_status",
|
||||||
party: status.party,
|
party: status.party,
|
||||||
members: status.members,
|
members: status.members,
|
||||||
|
|
@ -174,18 +162,7 @@ export const partyApp = new Elysia()
|
||||||
});
|
});
|
||||||
if (!result) return { party: null, members: [] };
|
if (!result) return { party: null, members: [] };
|
||||||
const status = await getPartyStatus(result.partyId);
|
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);
|
broadcastSnapshot(result.partyId, status);
|
||||||
reassignUserSocketsToParty(user.id, null);
|
|
||||||
return status ?? { party: null, members: [] };
|
return status ?? { party: null, members: [] };
|
||||||
},
|
},
|
||||||
{ auth: true },
|
{ auth: true },
|
||||||
|
|
@ -226,13 +203,7 @@ export const partyApp = new Elysia()
|
||||||
await cleanupPartyIfEmpty(tx, currentMembership.partyId);
|
await cleanupPartyIfEmpty(tx, currentMembership.partyId);
|
||||||
});
|
});
|
||||||
const status = await getPartyStatus(currentMembership.partyId);
|
const status = await getPartyStatus(currentMembership.partyId);
|
||||||
broadcastPartyEvent(currentMembership.partyId, {
|
|
||||||
type: "member_left",
|
|
||||||
userId: body.memberUserId,
|
|
||||||
kickedBy: user.id,
|
|
||||||
});
|
|
||||||
broadcastSnapshot(currentMembership.partyId, status);
|
broadcastSnapshot(currentMembership.partyId, status);
|
||||||
reassignUserSocketsToParty(body.memberUserId, null);
|
|
||||||
return status ?? { party: null, members: [] };
|
return status ?? { party: null, members: [] };
|
||||||
},
|
},
|
||||||
{
|
{
|
||||||
|
|
|
||||||
|
|
@ -1,4 +1,7 @@
|
||||||
import { useRouteContext } from "@tanstack/react-router";
|
import { useRouteContext } from "@tanstack/react-router";
|
||||||
|
import { useParty } from "#/hooks/use-party";
|
||||||
|
import { useUser } from "#/hooks/user";
|
||||||
|
import { initials } from "#/lib/utils";
|
||||||
import { Avatar, AvatarFallback, AvatarImage } from "./ui/avatar";
|
import { Avatar, AvatarFallback, AvatarImage } from "./ui/avatar";
|
||||||
import {
|
import {
|
||||||
Item,
|
Item,
|
||||||
|
|
@ -7,11 +10,10 @@ import {
|
||||||
ItemMedia,
|
ItemMedia,
|
||||||
ItemTitle,
|
ItemTitle,
|
||||||
} from "./ui/item";
|
} from "./ui/item";
|
||||||
import { useUser } from "#/hooks/user";
|
|
||||||
import { initials } from "#/lib/utils";
|
|
||||||
|
|
||||||
export function UserInfo() {
|
export function UserInfo() {
|
||||||
const { user } = useUser();
|
const { user } = useUser();
|
||||||
|
const { party, members, isConnecting, isReconnecting } = useParty();
|
||||||
return (
|
return (
|
||||||
<Item>
|
<Item>
|
||||||
<ItemMedia>
|
<ItemMedia>
|
||||||
|
|
@ -22,7 +24,15 @@ export function UserInfo() {
|
||||||
</ItemMedia>
|
</ItemMedia>
|
||||||
<ItemContent>
|
<ItemContent>
|
||||||
<ItemTitle>{user?.name}</ItemTitle>
|
<ItemTitle>{user?.name}</ItemTitle>
|
||||||
<ItemDescription>No party yet</ItemDescription>
|
<ItemDescription>
|
||||||
|
{isConnecting
|
||||||
|
? "Connecting..."
|
||||||
|
: isReconnecting
|
||||||
|
? "Reconnecting..."
|
||||||
|
: party
|
||||||
|
? `${members.length} in party`
|
||||||
|
: "No party yet"}
|
||||||
|
</ItemDescription>
|
||||||
</ItemContent>
|
</ItemContent>
|
||||||
</Item>
|
</Item>
|
||||||
);
|
);
|
||||||
|
|
|
||||||
127
web/src/hooks/use-party-socket.ts
Normal file
127
web/src/hooks/use-party-socket.ts
Normal file
|
|
@ -0,0 +1,127 @@
|
||||||
|
import { useCallback, useEffect, useMemo, useRef, useState } from "react";
|
||||||
|
import type { PartySocketEvent } from "../../../api/src/party-types";
|
||||||
|
|
||||||
|
type Handler = (event: PartySocketEvent) => void;
|
||||||
|
|
||||||
|
const PING_INTERVAL_MS = 30_000;
|
||||||
|
const RECONNECT_BASE_MS = 1_000;
|
||||||
|
const RECONNECT_MAX_MS = 30_000;
|
||||||
|
|
||||||
|
export function usePartySocket({
|
||||||
|
apiUrl,
|
||||||
|
onMessage,
|
||||||
|
}: {
|
||||||
|
apiUrl: string | null;
|
||||||
|
onMessage: Handler | null;
|
||||||
|
}) {
|
||||||
|
const [connectionState, setConnectionState] = useState<
|
||||||
|
"disconnected" | "connecting" | "connected" | "reconnecting"
|
||||||
|
>("disconnected");
|
||||||
|
|
||||||
|
const wsRef = useRef<WebSocket | null>(null);
|
||||||
|
const pingTimerRef = useRef<ReturnType<typeof setInterval> | null>(null);
|
||||||
|
const reconnectTimerRef = useRef<ReturnType<typeof setTimeout> | null>(null);
|
||||||
|
const reconnectAttemptRef = useRef(0);
|
||||||
|
const handlerRef = useRef(onMessage);
|
||||||
|
|
||||||
|
useEffect(() => {
|
||||||
|
handlerRef.current = onMessage;
|
||||||
|
}, [onMessage]);
|
||||||
|
|
||||||
|
const setupWs = useCallback(
|
||||||
|
(ws: WebSocket) => {
|
||||||
|
ws.onopen = () => {
|
||||||
|
reconnectAttemptRef.current = 0;
|
||||||
|
setConnectionState("connected");
|
||||||
|
pingTimerRef.current = setInterval(() => {
|
||||||
|
if (ws.readyState === WebSocket.OPEN) {
|
||||||
|
ws.send(JSON.stringify({ type: "ping" }));
|
||||||
|
}
|
||||||
|
}, PING_INTERVAL_MS);
|
||||||
|
};
|
||||||
|
|
||||||
|
ws.onmessage = (event) => {
|
||||||
|
const parsed = JSON.parse(event.data) as PartySocketEvent;
|
||||||
|
handlerRef.current?.(parsed);
|
||||||
|
};
|
||||||
|
|
||||||
|
ws.onclose = () => {
|
||||||
|
if (pingTimerRef.current) {
|
||||||
|
clearInterval(pingTimerRef.current);
|
||||||
|
pingTimerRef.current = null;
|
||||||
|
}
|
||||||
|
wsRef.current = null;
|
||||||
|
setConnectionState("reconnecting");
|
||||||
|
|
||||||
|
const delay = Math.min(
|
||||||
|
RECONNECT_BASE_MS * 2 ** reconnectAttemptRef.current,
|
||||||
|
RECONNECT_MAX_MS,
|
||||||
|
);
|
||||||
|
reconnectAttemptRef.current++;
|
||||||
|
reconnectTimerRef.current = setTimeout(() => {
|
||||||
|
if (!apiUrl) return;
|
||||||
|
const protocol = apiUrl.startsWith("https") ? "wss" : "ws";
|
||||||
|
const newWs = new WebSocket(
|
||||||
|
`${protocol}://${apiUrl.replace(/https?:\/\//, "")}/api/party-socket/ws`,
|
||||||
|
);
|
||||||
|
wsRef.current = newWs;
|
||||||
|
setupWs(newWs);
|
||||||
|
}, delay);
|
||||||
|
};
|
||||||
|
},
|
||||||
|
[apiUrl],
|
||||||
|
);
|
||||||
|
|
||||||
|
useEffect(() => {
|
||||||
|
if (!apiUrl) {
|
||||||
|
if (wsRef.current) {
|
||||||
|
wsRef.current.close();
|
||||||
|
wsRef.current = null;
|
||||||
|
}
|
||||||
|
if (pingTimerRef.current) {
|
||||||
|
clearInterval(pingTimerRef.current);
|
||||||
|
pingTimerRef.current = null;
|
||||||
|
}
|
||||||
|
if (reconnectTimerRef.current) {
|
||||||
|
clearTimeout(reconnectTimerRef.current);
|
||||||
|
reconnectTimerRef.current = null;
|
||||||
|
}
|
||||||
|
setConnectionState("disconnected");
|
||||||
|
reconnectAttemptRef.current = 0;
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
setConnectionState("connecting");
|
||||||
|
const protocol = apiUrl.startsWith("https") ? "wss" : "ws";
|
||||||
|
const ws = new WebSocket(
|
||||||
|
`${protocol}://${apiUrl.replace(/https?:\/\//, "")}/api/party-socket/ws`,
|
||||||
|
);
|
||||||
|
wsRef.current = ws;
|
||||||
|
setupWs(ws);
|
||||||
|
|
||||||
|
return () => {
|
||||||
|
ws.close();
|
||||||
|
wsRef.current = null;
|
||||||
|
if (pingTimerRef.current) {
|
||||||
|
clearInterval(pingTimerRef.current);
|
||||||
|
pingTimerRef.current = null;
|
||||||
|
}
|
||||||
|
if (reconnectTimerRef.current) {
|
||||||
|
clearTimeout(reconnectTimerRef.current);
|
||||||
|
reconnectTimerRef.current = null;
|
||||||
|
}
|
||||||
|
};
|
||||||
|
}, [apiUrl, setupWs]);
|
||||||
|
|
||||||
|
const state = useMemo(
|
||||||
|
() => ({
|
||||||
|
connectionState,
|
||||||
|
isConnected: connectionState === "connected",
|
||||||
|
isConnecting: connectionState === "connecting",
|
||||||
|
isReconnecting: connectionState === "reconnecting",
|
||||||
|
}),
|
||||||
|
[connectionState],
|
||||||
|
);
|
||||||
|
|
||||||
|
return state;
|
||||||
|
}
|
||||||
58
web/src/hooks/use-party.ts
Normal file
58
web/src/hooks/use-party.ts
Normal file
|
|
@ -0,0 +1,58 @@
|
||||||
|
import { useCallback, useMemo, useState } from "react";
|
||||||
|
import type {
|
||||||
|
PartyMember,
|
||||||
|
PartySocketEvent,
|
||||||
|
PartyState,
|
||||||
|
} from "../../../api/src/party-types";
|
||||||
|
import { usePartySocket } from "./use-party-socket";
|
||||||
|
import { useUser } from "./user";
|
||||||
|
|
||||||
|
function reducePartyState(
|
||||||
|
state: PartyState,
|
||||||
|
event: PartySocketEvent,
|
||||||
|
): PartyState {
|
||||||
|
switch (event.type) {
|
||||||
|
case "snapshot":
|
||||||
|
case "party_status":
|
||||||
|
return { party: event.party, members: event.members };
|
||||||
|
case "member_payload":
|
||||||
|
case "pong":
|
||||||
|
case "error":
|
||||||
|
return state;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
function getApiUrl(): string | null {
|
||||||
|
if (typeof window === "undefined") return null;
|
||||||
|
const envUrl = import.meta.env.VITE_BETTER_AUTH_URL;
|
||||||
|
if (envUrl) return envUrl;
|
||||||
|
return `${window.location.protocol}//${window.location.host}`;
|
||||||
|
}
|
||||||
|
|
||||||
|
export function useParty() {
|
||||||
|
const { session } = useUser();
|
||||||
|
const [state, setState] = useState<PartyState>({
|
||||||
|
party: null,
|
||||||
|
members: [],
|
||||||
|
});
|
||||||
|
|
||||||
|
const handleMessage = useCallback((event: PartySocketEvent) => {
|
||||||
|
setState((prev: PartyState) => reducePartyState(prev, event));
|
||||||
|
}, []);
|
||||||
|
|
||||||
|
const apiUrl = useMemo(() => {
|
||||||
|
const url = getApiUrl();
|
||||||
|
if (!url) return null;
|
||||||
|
return url;
|
||||||
|
}, []);
|
||||||
|
|
||||||
|
const wsState = usePartySocket({
|
||||||
|
apiUrl,
|
||||||
|
onMessage: session ? handleMessage : null,
|
||||||
|
});
|
||||||
|
|
||||||
|
return {
|
||||||
|
...state,
|
||||||
|
...wsState,
|
||||||
|
};
|
||||||
|
}
|
||||||
|
|
@ -20,7 +20,17 @@ const config = defineConfig({
|
||||||
],
|
],
|
||||||
server: {
|
server: {
|
||||||
proxy: {
|
proxy: {
|
||||||
"/api": "http://localhost:4000",
|
"/api": {
|
||||||
|
target: "http://localhost:4000",
|
||||||
|
changeOrigin: true,
|
||||||
|
rewrite: (path) =>
|
||||||
|
path.replace(/^\/api/, "/api"),
|
||||||
|
},
|
||||||
|
"/api/party-socket/ws": {
|
||||||
|
target: "ws://localhost:4000",
|
||||||
|
ws: true,
|
||||||
|
rewriteWsOrigin: true,
|
||||||
|
},
|
||||||
},
|
},
|
||||||
},
|
},
|
||||||
});
|
});
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue