aiolabs/lightning-pub
2
0
Fork 0
Code Issues 9 Pull requests 4 Projects Releases Packages Wiki Activity Actions

event log v3

Browse source
This commit is contained in:
boufni95 2024-06-25 19:32:17 +02:00
parent cafef8c248
commit 4325cdd07d
14 changed files with 80 additions and 38 deletions
Download patch file Download diff file Expand all files Collapse all files

12
datasource.js
View file

@ -4,22 +4,22 @@ import { UserReceivingInvoice } from "./build/src/services/storage/entity/UserRe
import { AddressReceivingTransaction } from "./build/src/services/storage/entity/AddressReceivingTransaction.js"
import { Application } from "./build/src/services/storage/entity/Application.js"
import { ApplicationUser } from "./build/src/services/storage/entity/ApplicationUser.js"
import { BalanceEvent } from "./build/src/services/storage/entity/BalanceEvent.js"
import { ChannelBalanceEvent } from "./build/src/services/storage/entity/ChannelsBalanceEvent.js"
import { Product } from "./build/src/services/storage/entity/Product.js"
import { RoutingEvent } from "./build/src/services/storage/entity/RoutingEvent.js"
import { UserBasicAuth } from "./build/src/services/storage/entity/UserBasicAuth.js"
import { UserEphemeralKey } from "./build/src/services/storage/entity/UserEphemeralKey.js"
import { UserInvoicePayment } from "./build/src/services/storage/entity/UserInvoicePayment.js"
import { UserReceivingAddress } from "./build/src/services/storage/entity/UserReceivingAddress.js"
import { UserToUserPayment } from "./build/src/services/storage/entity/UserToUserPayment.js"
import { UserTransactionPayment } from "./build/src/services/storage/entity/UserTransactionPayment.js"
import { LspOrder } from "./build/src/services/storage/entity/LspOrder.js"
import { Initial1703170309875 } from './build/src/services/storage/migrations/1703170309875-initial.js'
import { LspOrder1718387847693 } from './build/src/services/storage/migrations/1718387847693-lsp_order.js'
export default new DataSource({
type: "sqlite",
database: "source.sqlite",
database: "db.sqlite",
// logging: true,
migrations: [Initial1703170309875, LspOrder1718387847693],
entities: [User, UserReceivingInvoice, UserReceivingAddress, AddressReceivingTransaction, UserInvoicePayment, UserTransactionPayment,
UserBasicAuth, UserEphemeralKey, Product, UserToUserPayment, Application, ApplicationUser, UserToUserPayment, RoutingEvent, BalanceEvent, ChannelBalanceEvent],
UserBasicAuth, UserEphemeralKey, Product, UserToUserPayment, Application, ApplicationUser, UserToUserPayment, LspOrder],
// synchronize: true,
})

5
src/services/lnd/liquidityProvider.ts
View file

@ -11,6 +11,7 @@ export type LiquidityRequest = { action: 'spend' | 'receive', amount: number }
export type nostrCallback<T> = { startedAtMillis: number, type: 'single' | 'stream', f: (res: T) => void }
export class LiquidityProvider {
client: ReturnType<typeof newNostrClient>
clientCbs: Record<string, nostrCallback<any>> = {}
clientId: string = ""
@ -47,6 +48,10 @@ export class LiquidityProvider {
}, 1000)
}
GetProviderDestination() {
return this.pubDestination
}
AwaitProviderReady = async (): Promise<boolean> => {
if (!this.pubDestination) {
return false

6
src/services/lnd/lnd.ts
View file

@ -275,7 +275,8 @@ export default class {
const shouldUseLiquidityProvider = await this.ShouldUseLiquidityProvider({ action: 'receive', amount: value })
if (shouldUseLiquidityProvider || useProvider) {
const invoice = await this.liquidProvider.AddInvoice(value, memo)
return { payRequest: invoice }
const providerDst = this.liquidProvider.GetProviderDestination()
return { payRequest: invoice, providerDst }
}
const res = await this.lightning.addInvoice(AddInvoiceReq(value, expiry, false, memo), DeadLineMetadata())
this.log("new invoice", res.response.paymentRequest)
@ -310,7 +311,8 @@ export default class {
const shouldUseLiquidityProvider = await this.ShouldUseLiquidityProvider({ action: 'spend', amount })
if (shouldUseLiquidityProvider || useProvider) {
const res = await this.liquidProvider.PayInvoice(invoice)
return { feeSat: res.network_fee + res.service_fee, valueSat: res.amount_paid, paymentPreimage: res.preimage }
const providerDst = this.liquidProvider.GetProviderDestination()
return { feeSat: res.network_fee + res.service_fee, valueSat: res.amount_paid, paymentPreimage: res.preimage, providerDst }
}
const abortController = new AbortController()
const req = PayInvoiceReq(invoice, amount, feeLimit)

2
src/services/lnd/settings.ts
View file

@ -46,6 +46,7 @@ export type NodeInfo = {
}
export type Invoice = {
payRequest: string
providerDst?: string
}
export type DecodedInvoice = {
numSatoshis: number
@ -55,4 +56,5 @@ export type PaidInvoice = {
feeSat: number
valueSat: number
paymentPreimage: string
providerDst?: string
}

4
src/services/main/paymentManager.ts
View file

@ -126,7 +126,7 @@ export default class {
}
const use = await this.liquidityManager.beforeInvoiceCreation(req.amountSats)
const res = await this.lnd.NewInvoice(req.amountSats, req.memo, options.expiry, use === 'provider')
const userInvoice = await this.storage.paymentStorage.AddUserInvoice(user, res.payRequest, options)
const userInvoice = await this.storage.paymentStorage.AddUserInvoice(user, res.payRequest, options, res.providerDst)
const appId = options.linkedApplication ? options.linkedApplication.app_id : ""
this.storage.eventsLog.LogEvent({ type: 'new_invoice', userId: user.user_id, appUserId: "", appId, balance: user.balance_sats, data: userInvoice.invoice, amount: req.amountSats })
return {
@ -213,7 +213,7 @@ export default class {
this.log("refund routing fee", routingFeeLimit, payment.feeSat, "sats")
await this.storage.userStorage.IncrementUserBalance(userId, routingFeeLimit - payment.feeSat, "routing_fee_refund:" + invoice)
}
await this.storage.paymentStorage.UpdateExternalPayment(pendingPayment.serial_id, payment.feeSat, serviceFee, true)
await this.storage.paymentStorage.UpdateExternalPayment(pendingPayment.serial_id, payment.feeSat, serviceFee, true, payment.providerDst)
return { preimage: payment.paymentPreimage, amtPaid: payment.valueSat, networkFee: payment.feeSat, serialId: pendingPayment.serial_id }

20
src/services/main/sanityChecker.ts
View file

@ -18,8 +18,6 @@ export default class SanityChecker {
events: LoggedEvent[] = []
invoices: Invoice[] = []
payments: Payment[] = []
providerInvoices: Types.UserOperation[] = []
providerPayments: Types.UserOperation[] = []
incrementSources: Record<string, boolean> = {}
decrementSources: Record<string, boolean> = {}
decrementEvents: Record<string, { userId: string, refund: number, failure: boolean }> = {}
@ -110,16 +108,13 @@ export default class SanityChecker {
const refund = amt - (entry.paid_amount + entry.routing_fees + entry.service_fees)
this.decrementEvents[invoice] = { userId, refund, failure: false }
}
if (!entry.internal) {
if (!entry.internal && !entry.liquidityProvider) {
const lndEntry = this.payments.find(i => i.paymentRequest === invoice)
if (!lndEntry) {
const providerEntry = this.providerPayments.find(i => i.identifier === invoice)
if (!providerEntry) {
throw new Error("payment not found in lnd for invoice " + invoice)
}
}
}
}
async validateUser2UserPayment({ fromUser, toUser, serialId }: { fromUser: string, toUser: string, serialId?: number }) {
if (!serialId) {
@ -189,16 +184,13 @@ export default class SanityChecker {
if (entry.paid_at_unix <= 0) {
throw new Error("invoice not paid for invoice " + invoice)
}
if (!entry.internal) {
if (!entry.internal && !entry.liquidityProvider) {
const entry = this.invoices.find(i => i.paymentRequest === invoice)
if (!entry) {
const providerEntry = this.providerInvoices.find(i => i.identifier === invoice)
if (!providerEntry) {
throw new Error("invoice not found in lnd " + invoice)
}
}
}
}
async validateRoutingFeeRefund({ amt, invoice, userId }: { userId: string, invoice: string, amt: number }) {
const entry = this.decrementEvents[invoice]
@ -236,14 +228,6 @@ export default class SanityChecker {
this.events = await this.storage.eventsLog.GetAllLogs()
this.invoices = (await this.lnd.GetAllPaidInvoices(1000)).invoices
this.payments = (await this.lnd.GetAllPayments(1000)).payments
const providerUsable = await this.lnd.liquidProvider.AwaitProviderReady()
if (providerUsable) {
const ops = await this.lnd.liquidProvider.GetOperations()
this.providerInvoices = ops.latestIncomingInvoiceOperations.operations
this.providerPayments = ops.latestOutgoingInvoiceOperations.operations
} else {
this.log("provider not usable, skipping provider checks")
}
this.incrementSources = {}
this.decrementSources = {}

2
src/services/main/settings.ts
View file

@ -58,7 +58,7 @@ export const LoadMainSettingsFromEnv = (): MainSettings => {
}
export const LoadTestSettingsFromEnv = (): TestSettings => {
const eventLogPath = `logs/eventLogV2Test${Date.now()}.csv`
const eventLogPath = `logs/eventLogV3Test${Date.now()}.csv`
const settings = LoadMainSettingsFromEnv()
return {
...settings,

5
src/services/storage/entity/UserInvoicePayment.ts
View file

@ -34,6 +34,11 @@ export class UserInvoicePayment {
@ManyToOne(type => Application, { eager: true })
linkedApplication: Application | null
@Column({
nullable: true,
})
liquidityProvider?: string
@CreateDateColumn()
created_at: Date

5
src/services/storage/entity/UserReceivingInvoice.ts
View file

@ -58,6 +58,11 @@ export class UserReceivingInvoice {
})
zap_info?: ZapInfo
@Column({
nullable: true,
})
liquidityProvider?: string
@CreateDateColumn()
created_at: Date

2
src/services/storage/eventsLog.ts
View file

@ -1,7 +1,7 @@
import fs from 'fs'
import { parse, stringify } from 'csv'
import { getLogger } from '../helpers/logger.js'
//const eventLogPath = "logs/eventLogV2.csv"
//const eventLogPath = "logs/eventLogV3.csv"
type LoggedEventType = 'new_invoice' | 'new_address' | 'address_paid' | 'invoice_paid' | 'invoice_payment' | 'address_payment' | 'u2u_receiver' | 'u2u_sender' | 'balance_increment' | 'balance_decrement'
export type LoggedEvent = {
timestampMs: number

2
src/services/storage/index.ts
View file

@ -14,7 +14,7 @@ export type StorageSettings = {
dataDir: string
}
export const LoadStorageSettingsFromEnv = (): StorageSettings => {
return { dbSettings: LoadDbSettingsFromEnv(), eventLogPath: "logs/eventLogV2.csv", dataDir: process.env.DATA_DIR || "" }
return { dbSettings: LoadDbSettingsFromEnv(), eventLogPath: "logs/eventLogV3.csv", dataDir: process.env.DATA_DIR || "" }
}
export default class {
DB: DataSource | EntityManager

36
src/services/storage/migrations/1719335699480-liquidity_provider.ts Normal file
View file

@ -0,0 +1,36 @@
import { MigrationInterface, QueryRunner } from "typeorm";
export class LiquidityProvider1719335699480 implements MigrationInterface {
name = 'LiquidityProvider1719335699480'
public async up(queryRunner: QueryRunner): Promise<void> {
await queryRunner.query(`DROP INDEX "IDX_a131e6b58f084f1340538681b5"`);
await queryRunner.query(`CREATE TABLE "temporary_user_receiving_invoice" ("serial_id" integer PRIMARY KEY AUTOINCREMENT NOT NULL, "invoice" varchar NOT NULL, "expires_at_unix" integer NOT NULL, "paid_at_unix" integer NOT NULL DEFAULT (0), "internal" boolean NOT NULL DEFAULT (0), "paidByLnd" boolean NOT NULL DEFAULT (0), "callbackUrl" varchar NOT NULL DEFAULT (''), "paid_amount" integer NOT NULL DEFAULT (0), "service_fee" integer NOT NULL DEFAULT (0), "zap_info" text, "created_at" datetime NOT NULL DEFAULT (datetime('now')), "updated_at" datetime NOT NULL DEFAULT (datetime('now')), "userSerialId" integer, "productProductId" varchar, "payerSerialId" integer, "linkedApplicationSerialId" integer, "liquidityProvider" varchar, CONSTRAINT "FK_714a8b7d4f89f8a802ca181b789" FOREIGN KEY ("linkedApplicationSerialId") REFERENCES "application" ("serial_id") ON DELETE NO ACTION ON UPDATE NO ACTION, CONSTRAINT "FK_d4bb1e4c60e8a869f1f43ca2e31" FOREIGN KEY ("payerSerialId") REFERENCES "user" ("serial_id") ON DELETE NO ACTION ON UPDATE NO ACTION, CONSTRAINT "FK_5263bde2a519db9ea608b702ec8" FOREIGN KEY ("productProductId") REFERENCES "product" ("product_id") ON DELETE NO ACTION ON UPDATE NO ACTION, CONSTRAINT "FK_2c0dfb3483f3e5e7e3cdd5dc71f" FOREIGN KEY ("userSerialId") REFERENCES "user" ("serial_id") ON DELETE NO ACTION ON UPDATE NO ACTION)`);
await queryRunner.query(`INSERT INTO "temporary_user_receiving_invoice"("serial_id", "invoice", "expires_at_unix", "paid_at_unix", "internal", "paidByLnd", "callbackUrl", "paid_amount", "service_fee", "zap_info", "created_at", "updated_at", "userSerialId", "productProductId", "payerSerialId", "linkedApplicationSerialId") SELECT "serial_id", "invoice", "expires_at_unix", "paid_at_unix", "internal", "paidByLnd", "callbackUrl", "paid_amount", "service_fee", "zap_info", "created_at", "updated_at", "userSerialId", "productProductId", "payerSerialId", "linkedApplicationSerialId" FROM "user_receiving_invoice"`);
await queryRunner.query(`DROP TABLE "user_receiving_invoice"`);
await queryRunner.query(`ALTER TABLE "temporary_user_receiving_invoice" RENAME TO "user_receiving_invoice"`);
await queryRunner.query(`CREATE UNIQUE INDEX "IDX_a131e6b58f084f1340538681b5" ON "user_receiving_invoice" ("invoice") `);
await queryRunner.query(`DROP INDEX "IDX_a609a4d3d8d9b07b90692a3c45"`);
await queryRunner.query(`CREATE TABLE "temporary_user_invoice_payment" ("serial_id" integer PRIMARY KEY AUTOINCREMENT NOT NULL, "invoice" varchar NOT NULL, "paid_amount" integer NOT NULL, "routing_fees" integer NOT NULL, "service_fees" integer NOT NULL, "paid_at_unix" integer NOT NULL, "internal" boolean NOT NULL DEFAULT (0), "created_at" datetime NOT NULL DEFAULT (datetime('now')), "updated_at" datetime NOT NULL DEFAULT (datetime('now')), "userSerialId" integer, "linkedApplicationSerialId" integer, "liquidityProvider" varchar, CONSTRAINT "FK_6bcac90887eea1dc61d37db2994" FOREIGN KEY ("linkedApplicationSerialId") REFERENCES "application" ("serial_id") ON DELETE NO ACTION ON UPDATE NO ACTION, CONSTRAINT "FK_ef2aa6761ab681bbbd5f94e0fcb" FOREIGN KEY ("userSerialId") REFERENCES "user" ("serial_id") ON DELETE NO ACTION ON UPDATE NO ACTION)`);
await queryRunner.query(`INSERT INTO "temporary_user_invoice_payment"("serial_id", "invoice", "paid_amount", "routing_fees", "service_fees", "paid_at_unix", "internal", "created_at", "updated_at", "userSerialId", "linkedApplicationSerialId") SELECT "serial_id", "invoice", "paid_amount", "routing_fees", "service_fees", "paid_at_unix", "internal", "created_at", "updated_at", "userSerialId", "linkedApplicationSerialId" FROM "user_invoice_payment"`);
await queryRunner.query(`DROP TABLE "user_invoice_payment"`);
await queryRunner.query(`ALTER TABLE "temporary_user_invoice_payment" RENAME TO "user_invoice_payment"`);
await queryRunner.query(`CREATE UNIQUE INDEX "IDX_a609a4d3d8d9b07b90692a3c45" ON "user_invoice_payment" ("invoice") `);
}
public async down(queryRunner: QueryRunner): Promise<void> {
await queryRunner.query(`DROP INDEX "IDX_a609a4d3d8d9b07b90692a3c45"`);
await queryRunner.query(`ALTER TABLE "user_invoice_payment" RENAME TO "temporary_user_invoice_payment"`);
await queryRunner.query(`CREATE TABLE "user_invoice_payment" ("serial_id" integer PRIMARY KEY AUTOINCREMENT NOT NULL, "invoice" varchar NOT NULL, "paid_amount" integer NOT NULL, "routing_fees" integer NOT NULL, "service_fees" integer NOT NULL, "paid_at_unix" integer NOT NULL, "internal" boolean NOT NULL DEFAULT (0), "created_at" datetime NOT NULL DEFAULT (datetime('now')), "updated_at" datetime NOT NULL DEFAULT (datetime('now')), "userSerialId" integer, "linkedApplicationSerialId" integer, CONSTRAINT "FK_6bcac90887eea1dc61d37db2994" FOREIGN KEY ("linkedApplicationSerialId") REFERENCES "application" ("serial_id") ON DELETE NO ACTION ON UPDATE NO ACTION, CONSTRAINT "FK_ef2aa6761ab681bbbd5f94e0fcb" FOREIGN KEY ("userSerialId") REFERENCES "user" ("serial_id") ON DELETE NO ACTION ON UPDATE NO ACTION)`);
await queryRunner.query(`INSERT INTO "user_invoice_payment"("serial_id", "invoice", "paid_amount", "routing_fees", "service_fees", "paid_at_unix", "internal", "created_at", "updated_at", "userSerialId", "linkedApplicationSerialId") SELECT "serial_id", "invoice", "paid_amount", "routing_fees", "service_fees", "paid_at_unix", "internal", "created_at", "updated_at", "userSerialId", "linkedApplicationSerialId" FROM "temporary_user_invoice_payment"`);
await queryRunner.query(`DROP TABLE "temporary_user_invoice_payment"`);
await queryRunner.query(`CREATE UNIQUE INDEX "IDX_a609a4d3d8d9b07b90692a3c45" ON "user_invoice_payment" ("invoice") `);
await queryRunner.query(`DROP INDEX "IDX_a131e6b58f084f1340538681b5"`);
await queryRunner.query(`ALTER TABLE "user_receiving_invoice" RENAME TO "temporary_user_receiving_invoice"`);
await queryRunner.query(`CREATE TABLE "user_receiving_invoice" ("serial_id" integer PRIMARY KEY AUTOINCREMENT NOT NULL, "invoice" varchar NOT NULL, "expires_at_unix" integer NOT NULL, "paid_at_unix" integer NOT NULL DEFAULT (0), "internal" boolean NOT NULL DEFAULT (0), "paidByLnd" boolean NOT NULL DEFAULT (0), "callbackUrl" varchar NOT NULL DEFAULT (''), "paid_amount" integer NOT NULL DEFAULT (0), "service_fee" integer NOT NULL DEFAULT (0), "zap_info" text, "created_at" datetime NOT NULL DEFAULT (datetime('now')), "updated_at" datetime NOT NULL DEFAULT (datetime('now')), "userSerialId" integer, "productProductId" varchar, "payerSerialId" integer, "linkedApplicationSerialId" integer, CONSTRAINT "FK_714a8b7d4f89f8a802ca181b789" FOREIGN KEY ("linkedApplicationSerialId") REFERENCES "application" ("serial_id") ON DELETE NO ACTION ON UPDATE NO ACTION, CONSTRAINT "FK_d4bb1e4c60e8a869f1f43ca2e31" FOREIGN KEY ("payerSerialId") REFERENCES "user" ("serial_id") ON DELETE NO ACTION ON UPDATE NO ACTION, CONSTRAINT "FK_5263bde2a519db9ea608b702ec8" FOREIGN KEY ("productProductId") REFERENCES "product" ("product_id") ON DELETE NO ACTION ON UPDATE NO ACTION, CONSTRAINT "FK_2c0dfb3483f3e5e7e3cdd5dc71f" FOREIGN KEY ("userSerialId") REFERENCES "user" ("serial_id") ON DELETE NO ACTION ON UPDATE NO ACTION)`);
await queryRunner.query(`INSERT INTO "user_receiving_invoice"("serial_id", "invoice", "expires_at_unix", "paid_at_unix", "internal", "paidByLnd", "callbackUrl", "paid_amount", "service_fee", "zap_info", "created_at", "updated_at", "userSerialId", "productProductId", "payerSerialId", "linkedApplicationSerialId") SELECT "serial_id", "invoice", "expires_at_unix", "paid_at_unix", "internal", "paidByLnd", "callbackUrl", "paid_amount", "service_fee", "zap_info", "created_at", "updated_at", "userSerialId", "productProductId", "payerSerialId", "linkedApplicationSerialId" FROM "temporary_user_receiving_invoice"`);
await queryRunner.query(`DROP TABLE "temporary_user_receiving_invoice"`);
await queryRunner.query(`CREATE UNIQUE INDEX "IDX_a131e6b58f084f1340538681b5" ON "user_receiving_invoice" ("invoice") `);
}
}

3
src/services/storage/migrations/runner.ts
View file

@ -5,7 +5,8 @@ import { Initial1703170309875 } from './1703170309875-initial.js'
import { LndMetrics1703170330183 } from './1703170330183-lnd_metrics.js'
import { ChannelRouting1709316653538 } from './1709316653538-channel_routing.js'
import { LspOrder1718387847693 } from './1718387847693-lsp_order.js'
const allMigrations = [Initial1703170309875, LspOrder1718387847693]
import { LiquidityProvider1719335699480 } from './1719335699480-liquidity_provider.js'
const allMigrations = [Initial1703170309875, LspOrder1718387847693, LiquidityProvider1719335699480]
const allMetricsMigrations = [LndMetrics1703170330183, ChannelRouting1709316653538]
export const TypeOrmMigrationRunner = async (log: PubLogger, storageManager: Storage, settings: DbSettings, arg: string | undefined): Promise<boolean> => {
if (arg === 'fake_initial_migration') {

10
src/services/storage/paymentStorage.ts
View file

@ -92,7 +92,7 @@ export default class {
})
}
async AddUserInvoice(user: User, invoice: string, options: InboundOptionals = { expiry: defaultInvoiceExpiry }): Promise<UserReceivingInvoice> {
async AddUserInvoice(user: User, invoice: string, options: InboundOptionals = { expiry: defaultInvoiceExpiry }, providerDestination?: string): Promise<UserReceivingInvoice> {
const newUserInvoice = this.DB.getRepository(UserReceivingInvoice).create({
invoice: invoice,
callbackUrl: options.callbackUrl,
@ -101,7 +101,8 @@ export default class {
expires_at_unix: Math.floor(Date.now() / 1000) + options.expiry,
payer: options.expectedPayer,
linkedApplication: options.linkedApplication,
zap_info: options.zapInfo
zap_info: options.zapInfo,
liquidityProvider: providerDestination
})
return this.txQueue.PushToQueue<UserReceivingInvoice>({ exec: async db => db.getRepository(UserReceivingInvoice).save(newUserInvoice), dbTx: false, description: `add invoice for ${user.user_id} linked to ${options.linkedApplication?.app_id}: ${invoice} ` })
}
@ -167,11 +168,12 @@ export default class {
return this.txQueue.PushToQueue<UserInvoicePayment>({ exec: async db => db.getRepository(UserInvoicePayment).save(newPayment), dbTx: false, description: `add pending invoice payment for ${userId} linked to ${linkedApplication.app_id}: ${invoice}, amt: ${amount} ` })
}
async UpdateExternalPayment(invoicePaymentSerialId: number, routingFees: number, serviceFees: number, success: boolean) {
async UpdateExternalPayment(invoicePaymentSerialId: number, routingFees: number, serviceFees: number, success: boolean, providerDestination?: string) {
return this.DB.getRepository(UserInvoicePayment).update(invoicePaymentSerialId, {
routing_fees: routingFees,
service_fees: serviceFees,
paid_at_unix: success ? Math.floor(Date.now() / 1000) : -1
paid_at_unix: success ? Math.floor(Date.now() / 1000) : -1,
liquidityProvider: providerDestination
})
}