diff --git a/src/aimedb/db.ts b/src/aimedb/db.ts index cbccb36..9d48648 100644 --- a/src/aimedb/db.ts +++ b/src/aimedb/db.ts @@ -1,37 +1,34 @@ -import { PoolClient } from "pg"; -import sql from "sql-bricks"; +import sql from "sql-bricks-postgres"; -import { CardRepository, Transaction } from "./repo"; -import { connect, generateId, generateExtId } from "../db"; -import { AimeId } from "../model"; +import { CardRepository, Repositories } from "./repo"; +import { AimeId, generateExtId } from "../model"; +import { Transaction, generateId } from "../sql"; class CardRepositoryImpl implements CardRepository { - constructor(private readonly _conn: PoolClient) {} + constructor(private readonly _txn: Transaction) {} async lookup(luid: string, now: Date): Promise { const fetchSql = sql .select("c.id", "p.ext_id") .from("aime_card c") .join("aime_player p", { "c.player_id": "p.id" }) - .where("c.nfc_id", luid) - .toParams(); + .where("c.nfc_id", luid); - const { rows } = await this._conn.query(fetchSql); + const row = await this._txn.fetchRow(fetchSql); - if (rows.length === 0) { + if (row === undefined) { return undefined; } - const id = rows[0].id; - const extId = rows[0].ext_id; + const id = row.id; + const extId = row.ext_id; const touchSql = sql .update("aime_card") .set({ access_time: now }) - .where("id", id) - .toParams(); + .where("id", id); - await this._conn.query(touchSql); + await this._txn.modify(touchSql); return extId; } @@ -41,54 +38,32 @@ class CardRepositoryImpl implements CardRepository { const cardId = generateId(); const aimeId = generateExtId() as AimeId; - const playerSql = sql - .insert("aime_player", { - id: playerId, - ext_id: aimeId, - register_time: now, - }) - .toParams(); + const playerSql = sql.insert("aime_player", { + id: playerId, + ext_id: aimeId, + register_time: now, + }); - await this._conn.query(playerSql); + await this._txn.modify(playerSql); - const cardSql = sql - .insert("aime_card", { - id: cardId, - player_id: playerId, - nfc_id: luid, - register_time: now, - access_time: now, - }) - .toParams(); + const cardSql = sql.insert("aime_card", { + id: cardId, + player_id: playerId, + nfc_id: luid, + register_time: now, + access_time: now, + }); - await this._conn.query(cardSql); + await this._txn.modify(cardSql); return aimeId; } } -class TransactionImpl implements Transaction { - constructor(private readonly _conn: PoolClient) {} +export class SqlRepositories implements Repositories { + constructor(private readonly _txn: Transaction) {} cards(): CardRepository { - return new CardRepositoryImpl(this._conn); - } - - async commit(): Promise { - await this._conn.query("commit"); - await this._conn.release(); - } - - async rollback(): Promise { - await this._conn.query("rollback"); - await this._conn.release(); + return new CardRepositoryImpl(this._txn); } } - -export async function beginDbSession(): Promise { - const conn = await connect(); - - await conn.query("begin"); - - return new TransactionImpl(conn); -} diff --git a/src/aimedb/handler.ts b/src/aimedb/handler.ts index 84c135a..c93fbcc 100644 --- a/src/aimedb/handler.ts +++ b/src/aimedb/handler.ts @@ -3,6 +3,8 @@ import logger from "debug"; import { Repositories } from "./repo"; import * as Req from "./request"; import * as Res from "./response"; +import { Transaction } from "../sql"; +import { SqlRepositories } from "./db"; const debug = logger("app:aimedb:ops"); @@ -101,10 +103,12 @@ function log( } export async function dispatch( - rep: Repositories, + txn: Transaction, req: Req.AimeRequest, now: Date ): Promise { + const rep = new SqlRepositories(txn); + switch (req.type) { case "hello": return hello(rep, req, now); diff --git a/src/aimedb/index.ts b/src/aimedb/index.ts index 6debdc8..5538854 100644 --- a/src/aimedb/index.ts +++ b/src/aimedb/index.ts @@ -4,37 +4,37 @@ import { Socket } from "net"; import { dispatch } from "./handler"; import { AimeRequest } from "./request"; import { setup } from "./pipeline"; -import { beginDbSession } from "./db"; +import { DataSource } from "../sql/api"; const debug = logger("app:aimedb:session"); -export default async function aimedb(socket: Socket) { - debug("Connection opened"); +export default function aimedb(db: DataSource) { + return async function(socket: Socket) { + debug("Connection opened"); - const { input, output } = setup(socket); - const txn = await beginDbSession(); + const { input, output } = setup(socket); - try { for await (const obj of input) { - const now = new Date(); - const req = obj as AimeRequest; - const res = await dispatch(txn, req, now); + try { + const now = new Date(); + const req = obj as AimeRequest; + const res = await db.transaction(txn => dispatch(txn, req, now)); - if (res === undefined) { - debug("Closing connection"); + if (res === undefined) { + debug("Closing connection"); + + break; + } + + output.write(res); + } catch (e) { + debug(`Connection error:\n${e.toString()}\n`); break; } - - output.write(res); } - await txn.commit(); - } catch (e) { - debug(`Connection error:\n${e.toString()}\n`); - await txn.rollback(); - } - - debug("Connection closed"); - socket.end(); + debug("Connection closed"); + socket.end(); + }; } diff --git a/src/aimedb/repo.ts b/src/aimedb/repo.ts index 8deb52a..4c21a8c 100644 --- a/src/aimedb/repo.ts +++ b/src/aimedb/repo.ts @@ -9,9 +9,3 @@ export interface CardRepository { export interface Repositories { cards(): CardRepository; } - -export interface Transaction extends Repositories { - commit(): Promise; - - rollback(): Promise; -} diff --git a/src/db.ts b/src/db.ts deleted file mode 100644 index b666023..0000000 --- a/src/db.ts +++ /dev/null @@ -1,58 +0,0 @@ -import { randomBytes } from "crypto"; -import logger from "debug"; -import { Pool, PoolClient } from "pg"; - -export type Id = bigint & { __id: T }; - -const debug = logger("app:sql"); -const currentSchemaVer = 5; - -const pool = new Pool(); -const fence = testConnection(); - -export async function connect(): Promise { - await fence; - - return pool.connect(); -} - -export function generateId(): bigint { - const buf = randomBytes(8); - - buf[0] &= 0x7f; // Force number to be non-negative - - // Let's not depend on Node v12 for the sake of 3 LoC just yet. - - const hi = buf.readUInt32BE(0); - const lo = buf.readUInt32BE(4); - - return (BigInt(hi) << 32n) | BigInt(lo); -} - -/** Generate a random 32-bit ID for use in external protocol messages */ -export function generateExtId(): number { - const buf = randomBytes(4); - - buf[0] &= 0x7f; // Force number to be non-negative - - return buf.readUInt32BE(0); -} - -async function testConnection(): Promise { - const conn = await pool.connect(); - - try { - const { rows } = await conn.query("select schemaver from meta"); - const { schemaver } = rows[0]; - - if (schemaver !== currentSchemaVer) { - throw new Error( - `Expected schema version ${currentSchemaVer}, db is on ${schemaver}` - ); - } - - debug("Connection established"); - } finally { - conn.release(); - } -} diff --git a/src/idz/db/backgrounds.ts b/src/idz/db/backgrounds.ts index 48fd4dd..23e22df 100644 --- a/src/idz/db/backgrounds.ts +++ b/src/idz/db/backgrounds.ts @@ -1,24 +1,22 @@ -import { ClientBase } from "pg"; -import sql from "sql-bricks"; +import sql from "sql-bricks-postgres"; import { BackgroundCode } from "../model/base"; import { Profile } from "../model/profile"; import { FlagRepository } from "../repo"; -import { generateId, Id } from "../../db"; +import { Id, Transaction, generateId } from "../../sql"; export class SqlBackgroundsRepository implements FlagRepository { - constructor(private readonly _conn: ClientBase) {} + constructor(private readonly _txn: Transaction) {} async loadAll(id: Id): Promise> { const loadSql = sql .select("bg.background_no") .from("idz_background_unlock bg") .join("idz_profile p", { "bg.profile_id": "p.id" }) - .where("p.id", id) - .toParams(); + .where("p.id", id); - const { rows } = await this._conn.query(loadSql); + const rows = await this._txn.fetchRows(loadSql); const result = new Set(); for (const row of rows) { @@ -39,15 +37,13 @@ export class SqlBackgroundsRepository continue; } - const saveSql = sql - .insert("idz_background_unlock", { - id: generateId(), - profile_id: profileId, - background_no: flag, - }) - .toParams(); + const saveSql = sql.insert("idz_background_unlock", { + id: generateId(), + profile_id: profileId, + background_no: flag, + }); - await this._conn.query(saveSql); + await this._txn.modify(saveSql); } } } diff --git a/src/idz/db/car.ts b/src/idz/db/car.ts index 32ed78a..1f966c1 100644 --- a/src/idz/db/car.ts +++ b/src/idz/db/car.ts @@ -1,10 +1,9 @@ -import { ClientBase } from "pg"; import sql from "sql-bricks-postgres"; import { Car, CarSelector } from "../model/car"; import { Profile } from "../model/profile"; import { CarRepository } from "../repo"; -import { generateId, Id } from "../../db"; +import { Id, Transaction, generateId } from "../../sql"; function _extractRow(row: any): Car { return { @@ -27,29 +26,26 @@ function _extractRow(row: any): Car { } export class SqlCarRepository implements CarRepository { - constructor(private readonly _conn: ClientBase) {} + constructor(private readonly _txn: Transaction) {} async countCars(profileId: Id): Promise { const countSql = sql .select("count(*) result") .from("idz_car c") - .where("c.profile_id", profileId) - .toParams(); + .where("c.profile_id", profileId); - const { rows } = await this._conn.query(countSql); - const row = rows[0]; + const row = await this._txn.fetchRow(countSql); - return parseInt(row.result, 10); + return parseInt(row!.result, 10); } async loadAllCars(profileId: Id): Promise { const loadSql = sql .select("c.*") .from("idz_car c") - .where("c.profile_id", profileId) - .toParams(); + .where("c.profile_id", profileId); - const { rows } = await this._conn.query(loadSql); + const rows = await this._txn.fetchRows(loadSql); return rows.map(_extractRow); } @@ -59,12 +55,11 @@ export class SqlCarRepository implements CarRepository { .select("c.*") .from("idz_car c") .join("idz_car_selection s", { "c.id": "s.car_id" }) - .where("s.id", profileId) - .toParams(); + .where("s.id", profileId); - const { rows } = await this._conn.query(loadSql); + const row = await this._txn.fetchRow(loadSql); - return _extractRow(rows[0]); + return _extractRow(row); } async saveCar(profileId: Id, car: Car): Promise { @@ -104,10 +99,9 @@ export class SqlCarRepository implements CarRepository { "field_5b", "field_5c", "field_5e", - ]) - .toParams(); + ]); - await this._conn.query(saveSql); + await this._txn.modify(saveSql); } async saveSelection( @@ -118,11 +112,9 @@ export class SqlCarRepository implements CarRepository { .select("c.id") .from("idz_car c") .where("c.profile_id", profileId) - .where("c.selector", selector) - .toParams(); + .where("c.selector", selector); - const { rows } = await this._conn.query(findSql); - const row = rows[0]; + const row = await this._txn.fetchRow(findSql); if (row === undefined) { throw new Error("Selected car not found"); @@ -134,9 +126,8 @@ export class SqlCarRepository implements CarRepository { car_id: row.id, }) .onConflict("id") - .doUpdate(["car_id"]) - .toParams(); + .doUpdate(["car_id"]); - await this._conn.query(saveSql); + await this._txn.modify(saveSql); } } diff --git a/src/idz/db/chara.ts b/src/idz/db/chara.ts index 300d275..341dc7f 100644 --- a/src/idz/db/chara.ts +++ b/src/idz/db/chara.ts @@ -1,10 +1,9 @@ -import { ClientBase } from "pg"; import sql from "sql-bricks-postgres"; import { Chara } from "../model/chara"; import { Profile } from "../model/profile"; import { FacetRepository } from "../repo"; -import { Id } from "../../db"; +import { Id, Transaction } from "../../sql"; export function _extractChara(row: any): Chara { return { @@ -22,17 +21,15 @@ export function _extractChara(row: any): Chara { } export class SqlCharaRepository implements FacetRepository { - constructor(private readonly _conn: ClientBase) {} + constructor(private readonly _txn: Transaction) {} async load(profileId: Id): Promise { const loadSql = sql .select("c.*") .from("idz_chara c") - .where("c.id", profileId) - .toParams(); + .where("c.id", profileId); - const { rows } = await this._conn.query(loadSql); - const row = rows[0]; + const row = await this._txn.fetchRow(loadSql); return _extractChara(row); } @@ -63,9 +60,8 @@ export class SqlCharaRepository implements FacetRepository { "field_0e", "title", "background", - ]) - .toParams(); + ]); - await this._conn.query(saveSql); + await this._txn.modify(saveSql); } } diff --git a/src/idz/db/coursePlays.ts b/src/idz/db/coursePlays.ts index e61b2f9..f4ce414 100644 --- a/src/idz/db/coursePlays.ts +++ b/src/idz/db/coursePlays.ts @@ -1,22 +1,20 @@ -import { ClientBase } from "pg"; import sql from "sql-bricks-postgres"; -import { CourseNo, ExtId } from "../model/base"; +import { CourseNo } from "../model/base"; import { Profile } from "../model/profile"; import { CoursePlaysRepository } from "../repo"; -import { generateId, Id } from "../../db"; +import { Id, Transaction, generateId } from "../../sql"; export class SqlCoursePlaysRepository implements CoursePlaysRepository { - constructor(private readonly _conn: ClientBase) {} + constructor(private readonly _txn: Transaction) {} async loadAll(profileId: Id): Promise> { const loadSql = sql .select("cp.course_no", "cp.count") .from("idz_course_plays cp") - .where("cp.profile_id", profileId) - .toParams(); + .where("cp.profile_id", profileId); - const { rows } = await this._conn.query(loadSql); + const rows = await this._txn.fetchRows(loadSql); const result = new Map(); for (const row of rows) { @@ -39,10 +37,9 @@ export class SqlCoursePlaysRepository implements CoursePlaysRepository { count: v, }) .onConflict("profile_id", "course_no") - .doUpdate(["count"]) - .toParams(); + .doUpdate(["count"]); - await this._conn.query(saveSql); + await this._txn.modify(saveSql); } } } diff --git a/src/idz/db/index.ts b/src/idz/db/index.ts index ef21cc2..03ca617 100644 --- a/src/idz/db/index.ts +++ b/src/idz/db/index.ts @@ -1,5 +1,3 @@ -import { PoolClient } from "pg"; - import { SqlBackgroundsRepository } from "./backgrounds"; import { SqlCarRepository } from "./car"; import { SqlCharaRepository } from "./chara"; @@ -18,90 +16,72 @@ import { SqlTitlesRepository } from "./titles"; import { SqlUnlocksRepository } from "./unlocks"; import * as Model from "../model"; import * as Repo from "../repo"; -import { connect } from "../../db"; +import { Transaction } from "../../sql"; -class TransactionImpl implements Repo.Transaction { - constructor(private readonly _conn: PoolClient) {} +export class SqlRepositories implements Repo.Repositories { + constructor(private readonly _txn: Transaction) {} backgrounds(): Repo.FlagRepository { - return new SqlBackgroundsRepository(this._conn); + return new SqlBackgroundsRepository(this._txn); } car(): Repo.CarRepository { - return new SqlCarRepository(this._conn); + return new SqlCarRepository(this._txn); } chara(): Repo.FacetRepository { - return new SqlCharaRepository(this._conn); + return new SqlCharaRepository(this._txn); } coursePlays(): Repo.CoursePlaysRepository { - return new SqlCoursePlaysRepository(this._conn); + return new SqlCoursePlaysRepository(this._txn); } missions(): Repo.FacetRepository { - return new SqlMissionsRepository(this._conn); + return new SqlMissionsRepository(this._txn); } profile(): Repo.ProfileRepository { - return new SqlProfileRepository(this._conn); + return new SqlProfileRepository(this._txn); } settings(): Repo.FacetRepository { - return new SqlSettingsRepository(this._conn); + return new SqlSettingsRepository(this._txn); } story(): Repo.FacetRepository { - return new SqlStoryRepository(this._conn); + return new SqlStoryRepository(this._txn); } teams(): Repo.TeamRepository { - return new SqlTeamRepository(this._conn); + return new SqlTeamRepository(this._txn); } teamAuto(): Repo.TeamAutoRepository { - return new SqlTeamAutoRepository(this._conn); + return new SqlTeamAutoRepository(this._txn); } teamMembers(): Repo.TeamMemberRepository { - return new SqlTeamMemberRepository(this._conn); + return new SqlTeamMemberRepository(this._txn); } teamReservations(): Repo.TeamReservationRepository { - return new SqlTeamReservationRepository(this._conn); + return new SqlTeamReservationRepository(this._txn); } tickets(): Repo.FacetRepository { - return new SqlTicketsRepository(this._conn); + return new SqlTicketsRepository(this._txn); } timeAttack(): Repo.TimeAttackRepository { - return new SqlTimeAttackRepository(this._conn); + return new SqlTimeAttackRepository(this._txn); } titles(): Repo.FlagRepository { - return new SqlTitlesRepository(this._conn); + return new SqlTitlesRepository(this._txn); } unlocks(): Repo.FacetRepository { - return new SqlUnlocksRepository(this._conn); - } - - async commit(): Promise { - await this._conn.query("commit"); - await this._conn.release(); - } - - async rollback(): Promise { - await this._conn.query("rollback"); - await this._conn.release(); + return new SqlUnlocksRepository(this._txn); } } - -export async function beginDbSession(): Promise { - const conn = await connect(); - - await conn.query("begin"); - - return new TransactionImpl(conn); -} diff --git a/src/idz/db/missions.ts b/src/idz/db/missions.ts index 1ae01c1..44ccb08 100644 --- a/src/idz/db/missions.ts +++ b/src/idz/db/missions.ts @@ -1,13 +1,12 @@ -import { ClientBase } from "pg"; import sql from "sql-bricks-postgres"; import { MissionGrid, MissionState } from "../model/mission"; import { Profile } from "../model/profile"; import { FacetRepository } from "../repo"; -import { generateId, Id } from "../../db"; +import { Id, Transaction, generateId } from "../../sql"; export class SqlMissionsRepository implements FacetRepository { - constructor(private readonly _conn: ClientBase) {} + constructor(private readonly _txn: Transaction) {} async load(profileId: Id): Promise { const result: MissionState = { @@ -31,10 +30,9 @@ export class SqlMissionsRepository implements FacetRepository { const loadSoloSql = sql .select("sm.*") .from("idz_solo_mission_state sm") - .where("sm.profile_id", profileId) - .toParams(); + .where("sm.profile_id", profileId); - const { rows } = await this._conn.query(loadSoloSql); + const rows = await this._txn.fetchRows(loadSoloSql); for (const row of rows) { result.solo[row.grid_no].cells[row.cell_no] = row.value; @@ -64,10 +62,9 @@ export class SqlMissionsRepository implements FacetRepository { value: grid[j], }) .onConflict("profile_id", "grid_no", "cell_no") - .doUpdate(["value"]) - .toParams(); + .doUpdate(["value"]); - await this._conn.query(saveSql); + await this._txn.modify(saveSql); } } } diff --git a/src/idz/db/profile.ts b/src/idz/db/profile.ts index e59fbd0..cb6bd6a 100644 --- a/src/idz/db/profile.ts +++ b/src/idz/db/profile.ts @@ -1,10 +1,9 @@ -import sql from "sql-bricks"; -import { ClientBase } from "pg"; +import sql from "sql-bricks-postgres"; import { Profile } from "../model/profile"; import { ProfileRepository } from "../repo"; -import { generateId, Id } from "../../db"; import { AimeId } from "../../model"; +import { Id, Transaction, generateId } from "../../sql"; export function _extractProfile(row: any): Profile { return { @@ -21,7 +20,7 @@ export function _extractProfile(row: any): Profile { } export class SqlProfileRepository implements ProfileRepository { - constructor(private readonly _conn: ClientBase) {} + constructor(private readonly _txn: Transaction) {} async find(aimeId: AimeId): Promise> { const profileId = await this.peek(aimeId); @@ -38,11 +37,9 @@ export class SqlProfileRepository implements ProfileRepository { .select("p.id") .from("idz_profile p") .join("aime_player r", { "p.player_id": "r.id" }) - .where("r.ext_id", aimeId) - .toParams(); + .where("r.ext_id", aimeId); - const { rows } = await this._conn.query(lookupSql); - const row = rows[0]; + const row = await this._txn.fetchRow(lookupSql); if (row === undefined) { return undefined; @@ -56,12 +53,11 @@ export class SqlProfileRepository implements ProfileRepository { .select("p.*", "r.ext_id as aime_id") .from("idz_profile p") .join("aime_player r", { "p.player_id": "r.id" }) - .where("p.id", id) - .toParams(); + .where("p.id", id); - const { rows } = await this._conn.query(loadSql); + const row = await this._txn.fetchRow(loadSql); - return _extractProfile(rows[0]); + return _extractProfile(row); } async save(id: Id, profile: Profile): Promise { @@ -74,21 +70,18 @@ export class SqlProfileRepository implements ProfileRepository { mileage: profile.mileage, access_time: profile.accessTime, }) - .where("id", id) - .toParams(); + .where("id", id); - await this._conn.query(saveSql); + await this._txn.modify(saveSql); } async create(profile: Profile): Promise> { const findSql = sql .select("r.id") .from("aime_player r") - .where("r.ext_id", profile.aimeId) - .toParams(); + .where("r.ext_id", profile.aimeId); - const { rows } = await this._conn.query(findSql); - const row = rows[0]; + const row = await this._txn.fetchRow(findSql); if (row === undefined) { throw new Error("Aime ID not found"); @@ -97,22 +90,20 @@ export class SqlProfileRepository implements ProfileRepository { const id = generateId(); const playerId = row.id; - const createSql = sql - .insert("idz_profile", { - id: id, - player_id: playerId, - name: profile.name, - lv: profile.lv, - exp: profile.exp, - fame: profile.fame, - dpoint: profile.dpoint, - mileage: profile.mileage, - register_time: profile.registerTime, - access_time: profile.accessTime, - }) - .toParams(); + const createSql = sql.insert("idz_profile", { + id: id, + player_id: playerId, + name: profile.name, + lv: profile.lv, + exp: profile.exp, + fame: profile.fame, + dpoint: profile.dpoint, + mileage: profile.mileage, + register_time: profile.registerTime, + access_time: profile.accessTime, + }); - await this._conn.query(createSql); + await this._txn.modify(createSql); return id as Id; } diff --git a/src/idz/db/settings.ts b/src/idz/db/settings.ts index 073a5a0..5d95217 100644 --- a/src/idz/db/settings.ts +++ b/src/idz/db/settings.ts @@ -1,23 +1,24 @@ -import { ClientBase } from "pg"; import sql from "sql-bricks-postgres"; import { Settings } from "../model/settings"; import { Profile } from "../model/profile"; import { FacetRepository } from "../repo"; -import { Id } from "../../db"; +import { Id, Transaction } from "../../sql"; export class SqlSettingsRepository implements FacetRepository { - constructor(private readonly _conn: ClientBase) {} + constructor(private readonly _txn: Transaction) {} async load(profileId: Id): Promise { const loadSql = sql .select("s.*") .from("idz_settings s") - .where("s.id", profileId) - .toParams(); + .where("s.id", profileId); - const { rows } = await this._conn.query(loadSql); - const row = rows[0]; + const row = await this._txn.fetchRow(loadSql); + + if (row === undefined) { + throw new Error(`Settings not found, profileId=${profileId}`); + } return { music: row.music, @@ -37,9 +38,8 @@ export class SqlSettingsRepository implements FacetRepository { gauges: settings.gauges, }) .onConflict("id") - .doUpdate(["music", "pack", "paper_cup", "gauges"]) - .toParams(); + .doUpdate(["music", "pack", "paper_cup", "gauges"]); - await this._conn.query(saveSql); + await this._txn.modify(saveSql); } } diff --git a/src/idz/db/story.ts b/src/idz/db/story.ts index 6cfa096..b8e62db 100644 --- a/src/idz/db/story.ts +++ b/src/idz/db/story.ts @@ -1,23 +1,20 @@ -import { ClientBase } from "pg"; import sql from "sql-bricks-postgres"; import { Profile } from "../model/profile"; import { Story, StoryRow, StoryCell } from "../model/story"; import { FacetRepository } from "../repo"; -import { generateId, Id } from "../../db"; +import { Id, Transaction, generateId } from "../../sql"; export class SqlStoryRepository implements FacetRepository { - constructor(private readonly _conn: ClientBase) {} + constructor(private readonly _txn: Transaction) {} async load(profileId: Id): Promise { const loadSql = sql .select("s.*") .from("idz_story_state s") - .where("s.id", profileId) - .toParams(); + .where("s.id", profileId); - const headerResult = await this._conn.query(loadSql); - const header = headerResult.rows[0]; + const header = await this._txn.fetchRow(loadSql); // Must succeed even if nonexistent (required by save method below) @@ -40,10 +37,9 @@ export class SqlStoryRepository implements FacetRepository { const loadCellSql = sql .select("sc.*") .from("idz_story_cell_state sc") - .where("sc.profile_id", profileId) - .toParams(); + .where("sc.profile_id", profileId); - const { rows } = await this._conn.query(loadCellSql); + const rows = await this._txn.fetchRows(loadCellSql); for (const row of rows) { const cell = result.rows[row.row_no].cells[row.col_no]; @@ -65,10 +61,9 @@ export class SqlStoryRepository implements FacetRepository { y: story.y, }) .onConflict("id") - .doUpdate(["x", "y"]) - .toParams(); + .doUpdate(["x", "y"]); - await this._conn.query(headSql); + await this._txn.modify(headSql); for (let i = 0; i < story.rows.length; i++) { const exRow = existing.rows[i]; @@ -92,10 +87,9 @@ export class SqlStoryRepository implements FacetRepository { b: cell.b, }) .onConflict("profile_id", "row_no", "col_no") - .doUpdate(["a", "b"]) - .toParams(); + .doUpdate(["a", "b"]); - await this._conn.query(cellSql); + await this._txn.modify(cellSql); } } } diff --git a/src/idz/db/team.ts b/src/idz/db/team.ts index 85385ce..0fc5502 100644 --- a/src/idz/db/team.ts +++ b/src/idz/db/team.ts @@ -1,23 +1,21 @@ -import { ClientBase } from "pg"; -import sql from "sql-bricks"; +import sql from "sql-bricks-postgres"; import { ExtId } from "../model/base"; import { Team } from "../model/team"; import { TeamSpec, TeamRepository } from "../repo"; -import { Id, generateExtId, generateId } from "../../db"; +import { generateExtId } from "../../model"; +import { Id, Transaction, generateId } from "../../sql"; export class SqlTeamRepository implements TeamRepository { - constructor(private readonly _conn: ClientBase) {} + constructor(private readonly _txn: Transaction) {} async find(extId: ExtId): Promise> { const findSql = sql .select("t.id") .from("idz_team t") - .where("t.ext_id", extId) - .toParams(); + .where("t.ext_id", extId); - const { rows } = await this._conn.query(findSql); - const row = rows[0]; + const row = await this._txn.fetchRow(findSql); if (row === undefined) { throw new Error(`Team not found for ExtID ${extId}`); @@ -30,13 +28,11 @@ export class SqlTeamRepository implements TeamRepository { const loadSql = sql .select("t.*") .from("idz_team t") - .where("t.id", id) - .toParams(); + .where("t.id", id); - const { rows } = await this._conn.query(loadSql); - const row = rows[0]; + const row = await this._txn.fetchRow(loadSql); - if (row == undefined) { + if (row === undefined) { throw new Error("Team not found"); } @@ -55,38 +51,32 @@ export class SqlTeamRepository implements TeamRepository { name_bg: team.nameBg, name_fx: team.nameFx, }) - .where("id", id) - .toParams(); + .where("id", id); - await this._conn.query(saveSql); + await this._txn.modify(saveSql); } async create(team: TeamSpec): Promise<[Id, ExtId]> { const id = generateId() as Id; const extId = generateExtId() as ExtId; - const createSql = sql - .insert("idz_team", { - id: id, - ext_id: extId, - name: team.name, - name_bg: team.nameBg, - name_fx: team.nameFx, - register_time: team.registerTime, - }) - .toParams(); + const createSql = sql.insert("idz_team", { + id: id, + ext_id: extId, + name: team.name, + name_bg: team.nameBg, + name_fx: team.nameFx, + register_time: team.registerTime, + }); - await this._conn.query(createSql); + await this._txn.modify(createSql); return [id, extId]; } async delete(id: Id): Promise { - const deleteSql = sql - .delete("idz_team") - .where("id", id) - .toParams(); + const deleteSql = sql.delete("idz_team").where("id", id); - await this._conn.query(deleteSql); + await this._txn.modify(deleteSql); } } diff --git a/src/idz/db/teamAuto.ts b/src/idz/db/teamAuto.ts index b7cbd52..943d6a3 100644 --- a/src/idz/db/teamAuto.ts +++ b/src/idz/db/teamAuto.ts @@ -1,23 +1,20 @@ import sql from "sql-bricks-postgres"; -import { ClientBase } from "pg"; import { Team, TeamAuto } from "../model/team"; import { TeamAutoRepository } from "../repo"; -import { Id } from "../../db"; +import { Id, Transaction } from "../../sql"; export class SqlTeamAutoRepository implements TeamAutoRepository { - constructor(private readonly _conn: ClientBase) {} + constructor(private readonly _txn: Transaction) {} async peek(): Promise<[TeamAuto, Id] | undefined> { const peekSql = sql .select("tt.*") .from("idz_team_auto tt") .orderBy("serial_no desc", "name_idx desc") - .limit(1) - .toParams(); + .limit(1); - const { rows } = await this._conn.query(peekSql); - const row = rows[0]; + const row = await this._txn.fetchRow(peekSql); return ( row && [ @@ -31,14 +28,12 @@ export class SqlTeamAutoRepository implements TeamAutoRepository { } async push(teamId: Id, auto: TeamAuto): Promise { - const pushSql = sql - .insert("idz_team_auto", { - id: teamId, - serial_no: auto.serialNo, - name_idx: auto.nameIdx, - }) - .toParams(); + const pushSql = sql.insert("idz_team_auto", { + id: teamId, + serial_no: auto.serialNo, + name_idx: auto.nameIdx, + }); - await this._conn.query(pushSql); + await this._txn.modify(pushSql); } } diff --git a/src/idz/db/teamMember.ts b/src/idz/db/teamMember.ts index b2c25a3..532c29a 100644 --- a/src/idz/db/teamMember.ts +++ b/src/idz/db/teamMember.ts @@ -1,25 +1,22 @@ import sql from "sql-bricks-postgres"; -import { ClientBase } from "pg"; import { Profile } from "../model/profile"; import { Team, TeamMember } from "../model/team"; import { TeamMemberRepository } from "../repo"; -import { Id, generateId } from "../../db"; import { _extractProfile } from "./profile"; import { _extractChara } from "./chara"; +import { Id, Transaction } from "../../sql"; export class SqlTeamMemberRepository implements TeamMemberRepository { - constructor(private readonly _conn: ClientBase) {} + constructor(private readonly _txn: Transaction) {} async findTeam(profileId: Id): Promise | undefined> { const findSql = sql .select("tm.team_id") .from("idz_team_member tm") - .where("tm.id", profileId) - .toParams(); + .where("tm.id", profileId); - const { rows } = await this._conn.query(findSql); - const row = rows[0]; + const row = await this._txn.fetchRow(findSql); if (row === undefined) { return undefined; @@ -33,11 +30,9 @@ export class SqlTeamMemberRepository implements TeamMemberRepository { .select("tm.id") .from("idz_team_member tm") .where("tm.team_id", teamId) - .where("tm.leader", true) - .toParams(); + .where("tm.leader", true); - const { rows } = await this._conn.query(findSql); - const row = rows[0]; + const row = await this._txn.fetchRow(findSql); if (row === undefined) { return undefined; @@ -53,10 +48,9 @@ export class SqlTeamMemberRepository implements TeamMemberRepository { .join("idz_profile p", { "tm.id": "p.id" }) .join("idz_chara c", { "tm.id": "c.id" }) .join("aime_player r", { "p.player_id": "r.id" }) - .where("tm.team_id", teamId) - .toParams(); + .where("tm.team_id", teamId); - const { rows } = await this._conn.query(loadSql); + const rows = await this._txn.fetchRows(loadSql); return rows.map((row: any) => ({ profile: _extractProfile(row), @@ -77,10 +71,9 @@ export class SqlTeamMemberRepository implements TeamMemberRepository { .select("id") .from("idz_team") .where("id", teamId) - .forUpdate() - .toParams(); + .forUpdate(); - await this._conn.query(lockSql); + await this._txn.modify(lockSql); // Double-check (with lock held) that there is room to join this team. // If this fails then the error will propagate to the client and it will @@ -94,13 +87,11 @@ export class SqlTeamMemberRepository implements TeamMemberRepository { const countSql = sql .select("count(*) as count") .from("idz_team_member") - .where("team_id", teamId) - .toParams(); + .where("team_id", teamId); - const { rows } = await this._conn.query(countSql); - const row = rows[0]; + const row = await this._txn.fetchRow(countSql); - if (row.count >= 6) { + if (row!.count >= 6) { throw new Error(`Team ${teamId} is full`); } @@ -114,20 +105,18 @@ export class SqlTeamMemberRepository implements TeamMemberRepository { join_time: timestamp, }) .onConflict("id") - .doUpdate(["team_id", "leader", "join_time"]) - .toParams(); + .doUpdate(["team_id", "leader", "join_time"]); - await this._conn.query(joinSql); + await this._txn.modify(joinSql); } async leave(teamId: Id, profileId: Id): Promise { const leaveSql = sql .delete("idz_team_member") .where("team_id", teamId) - .where("id", profileId) - .toParams(); + .where("id", profileId); - await this._conn.query(leaveSql); + await this._txn.modify(leaveSql); } async makeLeader(teamId: Id, profileId: Id): Promise { @@ -135,19 +124,17 @@ export class SqlTeamMemberRepository implements TeamMemberRepository { .update("idz_team_member", { leader: false, }) - .where("team_id", teamId) - .toParams(); + .where("team_id", teamId); - await this._conn.query(clearSql); + await this._txn.modify(clearSql); const setSql = sql .update("idz_team_member", { leader: true, }) .where("id", profileId) - .where("team_id", teamId) - .toParams(); + .where("team_id", teamId); - await this._conn.query(setSql); + await this._txn.modify(setSql); } } diff --git a/src/idz/db/teamReservation.ts b/src/idz/db/teamReservation.ts index 247db2a..8e96406 100644 --- a/src/idz/db/teamReservation.ts +++ b/src/idz/db/teamReservation.ts @@ -1,24 +1,22 @@ import sql from "sql-bricks-postgres"; -import { ClientBase } from "pg"; import { Team } from "../model/team"; import { TeamReservationRepository } from "../repo"; -import { Id } from "../../db"; import { AimeId } from "../../model"; +import { Id, Transaction } from "../../sql"; export class SqlTeamReservationRepository implements TeamReservationRepository { - constructor(private readonly _conn: ClientBase) {} + constructor(private readonly _txn: Transaction) {} private async _lockTeam(teamId: Id): Promise { const lockSql = sql .select("t.id") .from("idz_team t") .where("t.id", teamId) - .forUpdate() - .toParams(); + .forUpdate(); - await this._conn.query(lockSql); + await this._txn.modify(lockSql); } async occupancyHack(teamId: Id): Promise { @@ -30,20 +28,18 @@ export class SqlTeamReservationRepository const memberSql = sql .select("count(*) as count") .from("idz_team_member tm") - .where("tm.team_id", teamId) - .toParams(); + .where("tm.team_id", teamId); - const memberRes = await this._conn.query(memberSql); - const memberCount = parseInt(memberRes.rows[0].count, 10); + const memberRes = await this._txn.fetchRow(memberSql); + const memberCount = parseInt(memberRes!.count, 10); const reservSql = sql .select("count(*) as count") .from("idz_team_reservation tr") - .where("tr.team_id", teamId) - .toParams(); + .where("tr.team_id", teamId); - const reservRes = await this._conn.query(reservSql); - const reservCount = parseInt(reservRes.rows[0].count, 10); + const reservRes = await this._txn.fetchRow(reservSql); + const reservCount = parseInt(reservRes!.count, 10); return memberCount + reservCount; } @@ -57,11 +53,9 @@ export class SqlTeamReservationRepository const lookupSql = sql .select("r.id") .from("aime_player r") - .where("r.ext_id", aimeId) - .toParams(); + .where("r.ext_id", aimeId); - const { rows } = await this._conn.query(lookupSql); - const row = rows[0]; + const row = await this._txn.fetchRow(lookupSql); if (row === undefined) { throw new Error(`Unknown Aime ID ${aimeId}`); @@ -77,10 +71,9 @@ export class SqlTeamReservationRepository leader: leader === "leader", }) .onConflict("id") - .doUpdate(["team_id"]) - .toParams(); + .doUpdate(["team_id"]); - await this._conn.query(insertSql); + await this._txn.modify(insertSql); } async commitHack(aimeId: AimeId): Promise { @@ -89,32 +82,25 @@ export class SqlTeamReservationRepository .from("idz_profile p") .join("aime_player r", { "p.player_id": "r.id" }) .join("idz_team_reservation tr", { "r.id": "tr.id" }) - .where("r.ext_id", aimeId) - .toParams(); + .where("r.ext_id", aimeId); - const { rows } = await this._conn.query(lookupSql); - const row = rows[0]; + const row = await this._txn.fetchRow(lookupSql); if (row === undefined) { throw new Error(`Reservation not found for Aime ID ${aimeId}`); } - const insertSql = sql - .insert("idz_team_member", { - id: row.profile_id, - team_id: row.team_id, - join_time: row.join_time, - leader: row.leader, - }) - .toParams(); + const insertSql = sql.insert("idz_team_member", { + id: row.profile_id, + team_id: row.team_id, + join_time: row.join_time, + leader: row.leader, + }); - await this._conn.query(insertSql); + await this._txn.modify(insertSql); - const cleanupSql = sql - .delete("idz_team_reservation") - .where("id", row.id) - .toParams(); + const cleanupSql = sql.delete("idz_team_reservation").where("id", row.id); - await this._conn.query(cleanupSql); + await this._txn.modify(cleanupSql); } } diff --git a/src/idz/db/tickets.ts b/src/idz/db/tickets.ts index b6cf29d..cf3aa65 100644 --- a/src/idz/db/tickets.ts +++ b/src/idz/db/tickets.ts @@ -1,25 +1,22 @@ -import { ClientBase } from "pg"; import sql from "sql-bricks-postgres"; import { Profile } from "../model/profile"; import { Tickets } from "../model/tickets"; import { FacetRepository } from "../repo"; -import { Id } from "../../db"; +import { Id, Transaction } from "../../sql"; // TODO free continue export class SqlTicketsRepository implements FacetRepository { - constructor(private readonly _conn: ClientBase) {} + constructor(private readonly _txn: Transaction) {} async load(profileId: Id): Promise { const loadSql = sql .select("fc.*") .from("idz_free_car fc") - .where("fc.id", profileId) - .toParams(); + .where("fc.id", profileId); - const { rows } = await this._conn.query(loadSql); - const row = rows[0]; + const row = await this._txn.fetchRow(loadSql); return { freeCar: row && { @@ -32,12 +29,9 @@ export class SqlTicketsRepository implements FacetRepository { const { freeCar } = tickets; if (!freeCar) { - const delSql = sql - .delete("idz_free_car") - .where("id", profileId) - .toParams(); + const delSql = sql.delete("idz_free_car").where("id", profileId); - await this._conn.query(delSql); + await this._txn.modify(delSql); } else { const saveSql = sql .insert("idz_free_car", { @@ -45,10 +39,9 @@ export class SqlTicketsRepository implements FacetRepository { valid_from: freeCar.validFrom, }) .onConflict("id") - .doUpdate(["valid_from"]) - .toParams(); + .doUpdate(["valid_from"]); - await this._conn.query(saveSql); + await this._txn.modify(saveSql); } } } diff --git a/src/idz/db/timeAttack.ts b/src/idz/db/timeAttack.ts index 75061b0..8b249c4 100644 --- a/src/idz/db/timeAttack.ts +++ b/src/idz/db/timeAttack.ts @@ -1,14 +1,13 @@ -import { ClientBase } from "pg"; import sql from "sql-bricks-postgres"; import { RouteNo } from "../model/base"; import { Profile } from "../model/profile"; import { TimeAttackScore } from "../model/timeAttack"; import { TimeAttackRepository, TopTenResult } from "../repo"; -import { generateId, Id } from "../../db"; +import { Id, Transaction, generateId } from "../../sql"; export class SqlTimeAttackRepository implements TimeAttackRepository { - constructor(private readonly _conn: ClientBase) {} + constructor(private readonly _txn: Transaction) {} async loadTopTen( routeNo: RouteNo, @@ -21,10 +20,9 @@ export class SqlTimeAttackRepository implements TimeAttackRepository { .where("ta.route_no", routeNo) .where(sql.gt("ta.timestamp", minTimestamp)) .orderBy(["ta.total_time asc", "ta.timestamp asc"]) - .limit(10) - .toParams(); + .limit(10); - const { rows } = await this._conn.query(loadSql); + const rows = await this._txn.fetchRows(loadSql); return rows.map(row => ({ driverName: row.name, @@ -44,10 +42,9 @@ export class SqlTimeAttackRepository implements TimeAttackRepository { const loadSql = sql .select("ta.*") .from("idz_ta_best ta") - .where("ta.profile_id", profileId) - .toParams(); + .where("ta.profile_id", profileId); - const { rows } = await this._conn.query(loadSql); + const rows = await this._txn.fetchRows(loadSql); return rows.map(row => ({ routeNo: row.route_no, @@ -61,8 +58,30 @@ export class SqlTimeAttackRepository implements TimeAttackRepository { } async save(profileId: Id, score: TimeAttackScore): Promise { - const logSql = sql - .insert("idz_ta_result", { + const logSql = sql.insert("idz_ta_result", { + id: generateId(), + profile_id: profileId, + route_no: score.routeNo, + total_time: score.totalTime, + section_times: score.sectionTimes, + flags: score.flags, + grade: score.grade, + car_selector: score.carSelector, + timestamp: score.timestamp, + }); + + await this._txn.modify(logSql); + + const existSql = sql + .select("ta.total_time") + .from("idz_ta_best ta") + .where("ta.profile_id", profileId) + .where("ta.route_no", score.routeNo); + + const row = await this._txn.fetchRow(existSql); + + if (row === undefined) { + const insertSql = sql.insert("idz_ta_best", { id: generateId(), profile_id: profileId, route_no: score.routeNo, @@ -72,37 +91,9 @@ export class SqlTimeAttackRepository implements TimeAttackRepository { grade: score.grade, car_selector: score.carSelector, timestamp: score.timestamp, - }) - .toParams(); + }); - await this._conn.query(logSql); - - const existSql = sql - .select("ta.total_time") - .from("idz_ta_best ta") - .where("ta.profile_id", profileId) - .where("ta.route_no", score.routeNo) - .toParams(); - - const { rows } = await this._conn.query(existSql); - const row = rows[0]; - - if (row === undefined) { - const insertSql = sql - .insert("idz_ta_best", { - id: generateId(), - profile_id: profileId, - route_no: score.routeNo, - total_time: score.totalTime, - section_times: score.sectionTimes, - flags: score.flags, - grade: score.grade, - car_selector: score.carSelector, - timestamp: score.timestamp, - }) - .toParams(); - - await this._conn.query(insertSql); + await this._txn.modify(insertSql); } else if (score.totalTime < row.total_time) { const updateSql = sql .update("idz_ta_best", { @@ -114,10 +105,9 @@ export class SqlTimeAttackRepository implements TimeAttackRepository { timestamp: score.timestamp, }) .where("profile_id", profileId) - .where("route_no", score.routeNo) - .toParams(); + .where("route_no", score.routeNo); - await this._conn.query(updateSql); + await this._txn.modify(updateSql); } } } diff --git a/src/idz/db/titles.ts b/src/idz/db/titles.ts index 0f7e9a6..81e478c 100644 --- a/src/idz/db/titles.ts +++ b/src/idz/db/titles.ts @@ -1,22 +1,20 @@ -import { ClientBase } from "pg"; -import sql from "sql-bricks"; +import sql from "sql-bricks-postgres"; -import { TitleCode, ExtId } from "../model/base"; +import { TitleCode } from "../model/base"; import { Profile } from "../model/profile"; import { FlagRepository } from "../repo"; -import { generateId, Id } from "../../db"; +import { Id, Transaction, generateId } from "../../sql"; export class SqlTitlesRepository implements FlagRepository { - constructor(private readonly _conn: ClientBase) {} + constructor(private readonly _txn: Transaction) {} async loadAll(profileId: Id): Promise> { const loadSql = sql .select("t.title_no") .from("idz_title_unlock t") - .where("t.profile_id", profileId) - .toParams(); + .where("t.profile_id", profileId); - const { rows } = await this._conn.query(loadSql); + const rows = await this._txn.fetchRows(loadSql); const result = new Set(); for (const row of rows) { @@ -34,15 +32,13 @@ export class SqlTitlesRepository implements FlagRepository { continue; } - const saveSql = sql - .insert("idz_title_unlock", { - id: generateId(), - profile_id: profileId, - title_no: flag, - }) - .toParams(); + const saveSql = sql.insert("idz_title_unlock", { + id: generateId(), + profile_id: profileId, + title_no: flag, + }); - await this._conn.query(saveSql); + await this._txn.modify(saveSql); } } } diff --git a/src/idz/db/unlocks.ts b/src/idz/db/unlocks.ts index cd8bb03..363911b 100644 --- a/src/idz/db/unlocks.ts +++ b/src/idz/db/unlocks.ts @@ -1,24 +1,24 @@ -import { ClientBase } from "pg"; import sql from "sql-bricks-postgres"; -import { ExtId } from "../model/base"; import { Profile } from "../model/profile"; import { Unlocks } from "../model/unlocks"; import { FacetRepository } from "../repo"; -import { Id } from "../../db"; +import { Id, Transaction } from "../../sql"; export class SqlUnlocksRepository implements FacetRepository { - constructor(private readonly _conn: ClientBase) {} + constructor(private readonly _txn: Transaction) {} async load(profileId: Id): Promise { const loadSql = sql .select("u.*") .from("idz_unlocks u") - .where("u.id", profileId) - .toParams(); + .where("u.id", profileId); - const { rows } = await this._conn.query(loadSql); - const row = rows[0]; + const row = await this._txn.fetchRow(loadSql); + + if (row === undefined) { + throw new Error(`Unlocks not found, profileId=${profileId}`); + } return { cup: row.cup, @@ -38,9 +38,8 @@ export class SqlUnlocksRepository implements FacetRepository { last_mileage_reward: unlocks.lastMileageReward, }) .onConflict("id") - .doUpdate(["cup", "gauges", "music", "last_mileage_reward"]) - .toParams(); + .doUpdate(["cup", "gauges", "music", "last_mileage_reward"]); - await this._conn.query(saveSql); + await this._txn.modify(saveSql); } } diff --git a/src/idz/handler/_team.ts b/src/idz/handler/_team.ts index 57fdea4..07685e1 100644 --- a/src/idz/handler/_team.ts +++ b/src/idz/handler/_team.ts @@ -1,6 +1,6 @@ import { Team } from "../model/team"; import { Repositories } from "../repo"; -import { Id } from "../../db"; +import { Id } from "../../sql"; // Bleh. This factorization is kind of messy. diff --git a/src/idz/handler/index.ts b/src/idz/handler/index.ts index 68224b9..293770e 100644 --- a/src/idz/handler/index.ts +++ b/src/idz/handler/index.ts @@ -43,11 +43,15 @@ import { updateUserLog } from "./updateUserLog"; import { Request } from "../request"; import { Response } from "../response"; import { Repositories } from "../repo"; +import { Transaction } from "../../sql"; +import { SqlRepositories } from "../db"; export async function dispatch( - w: Repositories, + txn: Transaction, req: Request ): Promise { + const w = new SqlRepositories(txn); + switch (req.type) { case "check_team_name_req": return checkTeamName(w, req); diff --git a/src/idz/index.ts b/src/idz/index.ts index 830f431..9439f93 100644 --- a/src/idz/index.ts +++ b/src/idz/index.ts @@ -1,30 +1,28 @@ import logger from "debug"; import { Socket } from "net"; -import { beginDbSession } from "./db"; import { dispatch } from "./handler"; import { setup } from "./setup"; +import { DataSource } from "../sql"; const debug = logger("app:idz:session"); -export default async function idz(socket: Socket) { - const txn = await beginDbSession(); - const { input, output } = setup(socket); +export default function idz(db: DataSource) { + return async function(socket: Socket) { + const { input, output } = setup(socket); - debug("Connection opened"); + debug("Connection opened"); - try { - for await (const req of input) { - output.write(await dispatch(txn, req)); + try { + for await (const req of input) { + output.write(await db.transaction(txn => dispatch(txn, req))); + } + } catch (e) { + debug(`Error:\n${e.toString()}`); } - await txn.commit(); - } catch (e) { - debug(`Error:\n${e.toString()}`); - await txn.rollback(); - } + debug("Connection closed"); - debug("Connection closed"); - - input.end(); + input.end(); + }; } diff --git a/src/idz/repo.ts b/src/idz/repo.ts index f578371..d1577b2 100644 --- a/src/idz/repo.ts +++ b/src/idz/repo.ts @@ -2,7 +2,9 @@ import { Subtract } from "utility-types"; import * as Model from "./model"; import { AimeId } from "../model"; -import { Id } from "../db"; +import { Id } from "../sql"; + +// Id<> is a layer break here... need to find a better way to deal with this. export type TeamSpec = Subtract< Model.Team, @@ -164,9 +166,3 @@ export interface Repositories { unlocks(): FacetRepository; } - -export interface Transaction extends Repositories { - commit(): Promise; - - rollback(): Promise; -} diff --git a/src/index.ts b/src/index.ts index de27431..591ee14 100644 --- a/src/index.ts +++ b/src/index.ts @@ -1,4 +1,5 @@ import "dotenv/config"; + import fs from "fs"; import https from "https"; import http from "http"; @@ -11,21 +12,24 @@ import chunithm from "./chunithm"; import diva from "./diva"; import idz from "./idz"; import idzPing from "./idz/ping"; +import { openDataSource } from "./sql"; import * as Swb from "./switchboard"; +const db = openDataSource(); + const tls = { cert: fs.readFileSync("pki/server.pem"), key: fs.readFileSync("pki/server.key"), }; -net.createServer(aimedb).listen(Swb.PORT_AIMEDB, Swb.HOST_INT); +net.createServer(aimedb(db)).listen(Swb.PORT_AIMEDB, Swb.HOST_INT); http.createServer(allnet).listen(Swb.PORT_ALLNET, Swb.HOST_INT); https.createServer(tls, billing).listen(Swb.PORT_BILLING, Swb.HOST_INT); http.createServer(chunithm).listen(Swb.PORT_CHUNITHM, Swb.HOST_INT); http.createServer(diva).listen(Swb.PORT_DIVA, Swb.HOST_INT); -net.createServer(idz).listen(Swb.PORT_IDZ.USERDB.TCP, Swb.HOST_INT); +net.createServer(idz(db)).listen(Swb.PORT_IDZ.USERDB.TCP, Swb.HOST_INT); idzPing(10001, Swb.HOST_INT); // ?? tbd idzPing(Swb.PORT_IDZ.MATCH.UDP_SEND, Swb.HOST_INT); idzPing(Swb.PORT_IDZ.ECHO1, Swb.HOST_INT); diff --git a/src/model.ts b/src/model.ts index 2d1bfd5..98ddf2a 100644 --- a/src/model.ts +++ b/src/model.ts @@ -1 +1,12 @@ +import { randomBytes } from "crypto"; + export type AimeId = number & { __aimeId: null }; + +/** Generate a random 32-bit ID for use in external protocol messages */ +export function generateExtId(): number { + const buf = randomBytes(4); + + buf[0] &= 0x7f; // Force number to be non-negative + + return buf.readUInt32BE(0); +} diff --git a/src/sql/api.ts b/src/sql/api.ts new file mode 100644 index 0000000..2dc8d4a --- /dev/null +++ b/src/sql/api.ts @@ -0,0 +1,19 @@ +import * as sql from "sql-bricks-postgres"; + +export type Id = bigint & { __id: T }; + +export interface Row { + [key: string]: any; +} + +export interface Transaction { + modify(stmt: sql.Statement): Promise; + + fetchRow(stmt: sql.SelectStatement): Promise; + + fetchRows(stmt: sql.SelectStatement): Promise; +} + +export interface DataSource { + transaction(callback: (txn: Transaction) => Promise): Promise; +} diff --git a/src/sql/index.ts b/src/sql/index.ts new file mode 100644 index 0000000..fc5c101 --- /dev/null +++ b/src/sql/index.ts @@ -0,0 +1,3 @@ +export * from "./api"; +export * from "./pg"; +export * from "./util"; diff --git a/src/sql/pg.ts b/src/sql/pg.ts new file mode 100644 index 0000000..1e56c25 --- /dev/null +++ b/src/sql/pg.ts @@ -0,0 +1,57 @@ +import { Pool, PoolClient } from "pg"; +import * as sql from "sql-bricks-postgres"; + +import { DataSource, Row, Transaction } from "./api"; + +class PgTransaction implements Transaction { + constructor(private readonly _conn: PoolClient) {} + + async modify(stmt: sql.Statement): Promise { + await this._conn.query(stmt.toParams()); + } + + async fetchRow(stmt: sql.SelectStatement): Promise { + const { rows } = await this._conn.query(stmt.toParams()); + + return rows[0]; + } + + async fetchRows(stmt: sql.SelectStatement): Promise { + const { rows } = await this._conn.query(stmt.toParams()); + + return rows; + } +} + +class PgDataSource implements DataSource { + private readonly _pool: Pool; + + constructor() { + this._pool = new Pool(); + } + + async transaction( + callback: (txn: Transaction) => Promise + ): Promise { + const conn = await this._pool.connect(); + + await conn.query("begin"); + + try { + const txn = new PgTransaction(conn); + const result = await callback(txn); + + await conn.query("commit"); + + return result; + } catch (e) { + await conn.query("rollback"); + + return Promise.reject(e); + } + } +} + +export function openDataSource(): DataSource { + return new PgDataSource(); +} diff --git a/src/sql/util.ts b/src/sql/util.ts new file mode 100644 index 0000000..45c01bf --- /dev/null +++ b/src/sql/util.ts @@ -0,0 +1,14 @@ +import { randomBytes } from "crypto"; + +export function generateId(): bigint { + const buf = randomBytes(8); + + buf[0] &= 0x7f; // Force number to be non-negative + + // Let's not depend on Node v12 for the sake of 3 LoC just yet. + + const hi = buf.readUInt32BE(0); + const lo = buf.readUInt32BE(4); + + return (BigInt(hi) << 32n) | BigInt(lo); +}