Bridge-and-Join-s/internal/lkgateway/lkstore.go

package lkgateway

// Данные модуля личного кабинета клиента (bj-lk) в нашей PostgreSQL (база bj, схема lk):
// пользователи, позиции клиентов, заявки на перевод и итоги по ним. До 15.09.2026 лежали
// в базе обмена bj_exchange на MSSQL (схема «крана» Fancy); данные перенесены, «кран» убран.

import (
	"context"
	"database/sql"
	"encoding/json"
	"errors"
	"fmt"
	"strings"
	"sync"
	"time"

	"git.zetit.ru/zuevav/Bridge-and-Join-s/internal/m2m"
)

var (
	lkDBMu   sync.Mutex
	lkDBDSN  string
	lkDBConn *sql.DB
)

// lkDB — общий пул подключений к PostgreSQL bj (Postgres.DSN). Закрывать не нужно.
func lkDB(rc *RuntimeConfig) (*sql.DB, error) {
	dsn := strings.TrimSpace(rc.Snapshot().Postgres.DSN)
	if dsn == "" {
		return nil, errors.New("база PostgreSQL bj не настроена")
	}
	lkDBMu.Lock()
	defer lkDBMu.Unlock()
	if lkDBConn != nil && lkDBDSN == dsn {
		return lkDBConn, nil
	}
	if lkDBConn != nil {
		lkDBConn.Close()
	}
	db, err := sql.Open("pgx", dsn)
	if err != nil {
		return nil, err
	}
	db.SetMaxOpenConns(4)
	db.SetConnMaxLifetime(10 * time.Minute)
	lkDBConn, lkDBDSN = db, dsn
	return db, nil
}

// lkDBReady — PostgreSQL bj доступна (для /api/lk/ping).
func lkDBReady(rc *RuntimeConfig) bool {
	db, err := lkDB(rc)
	if err != nil {
		return false
	}
	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
	defer cancel()
	return db.PingContext(ctx) == nil
}

var lkSchemaDDL = []string{
	`CREATE SCHEMA IF NOT EXISTS lk`,
	`CREATE TABLE IF NOT EXISTS lk.users (
  id BIGSERIAL PRIMARY KEY, login VARCHAR(64) NOT NULL UNIQUE, password_hash VARCHAR(256) NOT NULL,
  contract_number VARCHAR(64) NOT NULL, full_name VARCHAR(128), last_name VARCHAR(64), first_name VARCHAR(64),
  middle_name VARCHAR(64), doc_type VARCHAR(8), doc_series VARCHAR(16), doc_number VARCHAR(32), email VARCHAR(128),
  phone VARCHAR(32), active BOOLEAN NOT NULL DEFAULT true, is_admin BOOLEAN NOT NULL DEFAULT false,
  can_transfer BOOLEAN NOT NULL DEFAULT false, created_at TIMESTAMPTZ NOT NULL DEFAULT now(), last_login_at TIMESTAMPTZ)`,
	`CREATE TABLE IF NOT EXISTS lk.client_positions (
  id BIGSERIAL PRIMARY KEY, contract_number VARCHAR(64) NOT NULL, client_name VARCHAR(128), isin VARCHAR(20),
  security_code VARCHAR(40), security_name VARCHAR(256), quantity NUMERIC(28,8), blocked NUMERIC(28,8),
  available NUMERIC(28,8), deponent_code VARCHAR(16), account_id VARCHAR(40), section_id VARCHAR(40),
  settlement_inn VARCHAR(16), updated_at TIMESTAMPTZ NOT NULL DEFAULT now())`,
	`CREATE INDEX IF NOT EXISTS ix_lk_positions_contract ON lk.client_positions(contract_number)`,
	`CREATE TABLE IF NOT EXISTS lk.transfer_requests (
  id BIGSERIAL PRIMARY KEY, created_at TIMESTAMPTZ NOT NULL DEFAULT now(), status VARCHAR(32) NOT NULL DEFAULT 'new',
  source VARCHAR(16), bj_claim_id VARCHAR(64), lk_request_id VARCHAR(64), contract_number VARCHAR(64),
  lk_investor_id VARCHAR(64), signed_document TEXT, signature_format VARCHAR(32), guid VARCHAR(40) NOT NULL,
  creation_timestamp VARCHAR(40), sender_code VARCHAR(16), receiver_code VARCHAR(16), cost_info VARCHAR(4),
  cost_code VARCHAR(16), investor_last VARCHAR(64), investor_first VARCHAR(64), investor_middle VARCHAR(64),
  doc_type VARCHAR(8), doc_series VARCHAR(16), doc_number VARCHAR(32), iia_agreement_type VARCHAR(8),
  iia_agreement_number VARCHAR(128), iia_agreement_date VARCHAR(16), iia_broker_inn VARCHAR(16),
  transferring_inn VARCHAR(16), receiving_inn VARCHAR(16), operation_type VARCHAR(8), request_xml TEXT, lk_payload TEXT)`,
	`CREATE INDEX IF NOT EXISTS ix_lk_requests_guid ON lk.transfer_requests(guid)`,
	`CREATE INDEX IF NOT EXISTS ix_lk_requests_contract ON lk.transfer_requests(contract_number)`,
	`CREATE TABLE IF NOT EXISTS lk.transfer_securities (
  id BIGSERIAL PRIMARY KEY, request_id BIGINT NOT NULL REFERENCES lk.transfer_requests(id), seq INT NOT NULL,
  reference_id VARCHAR(40), security_code VARCHAR(40), isin VARCHAR(20), sec_classification VARCHAR(32),
  sec_category VARCHAR(32), sec_type VARCHAR(64), sec_series VARCHAR(64), ident_reg_number VARCHAR(32),
  fund_reg_number VARCHAR(256), fund_class VARCHAR(128), quantity_whole BIGINT, quantity_fractional VARCHAR(40),
  settlement_inn VARCHAR(16), deponent_code VARCHAR(16), account_id VARCHAR(40), section_id VARCHAR(40),
  settlement_count INT, settlements_json TEXT, isolation_status VARCHAR(16))`,
	`CREATE INDEX IF NOT EXISTS ix_lk_securities_request ON lk.transfer_securities(request_id)`,
	`CREATE TABLE IF NOT EXISTS lk.transfer_responses (
  id BIGSERIAL PRIMARY KEY, request_id BIGINT, security_id BIGINT, guid VARCHAR(40),
  created_at TIMESTAMPTZ NOT NULL DEFAULT now(), verdict VARCHAR(32), reason VARCHAR(512), checked_by VARCHAR(64),
  check_source VARCHAR(64), securities_total INT, securities_ok INT, securities_failed INT, payload_json TEXT)`,
	`CREATE INDEX IF NOT EXISTS ix_lk_responses_request ON lk.transfer_responses(request_id)`,
}

// ensureLKSchema — схема lk и таблицы модуля ЛК (идемпотентно), при старте bj.
// lkSchemaExtra — таблицы, добавленные позже основной схемы.
func lkSchemaExtra() []string { return []string{lkDocSchema, lkAuditSchema} }

func ensureLKSchema(ctx context.Context, rc *RuntimeConfig) error {
	db, err := lkDB(rc)
	if err != nil {
		return err
	}
	cctx, cancel := context.WithTimeout(ctx, 30*time.Second)
	defer cancel()
	for _, q := range append(append([]string{}, lkSchemaDDL...), lkSchemaExtra()...) {
		if _, err := db.ExecContext(cctx, q); err != nil {
			return err
		}
	}
	return nil
}

// ClientPosition — остаток бумаги на счёте клиента (для Личного кабинета).
type ClientPosition struct {
	ISIN         string  `json:"isin"`
	SecurityCode string  `json:"security_code"`
	SecurityName string  `json:"security_name"`
	Quantity     float64 `json:"quantity"`
	Blocked      float64 `json:"blocked"`
	Available    float64 `json:"available"`
	DeponentCode string  `json:"deponent_code"`
	AccountID    string  `json:"account_id"`
	SectionID    string  `json:"section_id"`
	SettlementIN string  `json:"settlement_inn"`
	ClientName   string  `json:"client_name"`
	UpdatedAt    string  `json:"updated_at"`
	// Transferable — бумагу можно перевести через M2M (только российские выпуски).
	Transferable bool   `json:"transferable"`
	Note         string `json:"note"`
	// Оценка по данным торговой системы (пакет «Цены»); 0 — цены нет.
	Price     float64 `json:"price,omitempty"`
	PriceAt   string  `json:"price_board,omitempty"` // класс торгов, откуда цена
	Value     float64 `json:"value,omitempty"`       // стоимость позиции
	ChangePct float64 `json:"change_pct,omitempty"`  // изменение цены к закрытию, %
}

// lkClientPositions — позиции клиента модуля ЛК по номеру договора (тестовые клиенты;
// реальный портфель берётся из DEPO).
func lkClientPositions(ctx context.Context, rc *RuntimeConfig, contract string) ([]ClientPosition, error) {
	db, err := lkDB(rc)
	if err != nil {
		return nil, err
	}
	cctx, cancel := context.WithTimeout(ctx, 15*time.Second)
	defer cancel()
	rows, err := db.QueryContext(cctx, `SELECT COALESCE(isin,''), COALESCE(security_code,''), COALESCE(security_name,''),
		COALESCE(quantity,0), COALESCE(blocked,0), COALESCE(available,0),
		COALESCE(deponent_code,''), COALESCE(account_id,''), COALESCE(section_id,''),
		COALESCE(settlement_inn,''), COALESCE(client_name,''), to_char(updated_at,'DD.MM.YYYY HH24:MI')
		FROM lk.client_positions WHERE contract_number=$1 ORDER BY security_name, isin`, contract)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var out []ClientPosition
	for rows.Next() {
		var p ClientPosition
		if err := rows.Scan(&p.ISIN, &p.SecurityCode, &p.SecurityName, &p.Quantity, &p.Blocked, &p.Available,
			&p.DeponentCode, &p.AccountID, &p.SectionID, &p.SettlementIN, &p.ClientName, &p.UpdatedAt); err != nil {
			return out, err
		}
		out = append(out, p)
	}
	return out, rows.Err()
}

// LKRequestMeta — метаданные заявки модуля ЛК.
type LKRequestMeta struct {
	Source          string // 'lk'
	BJClaimID       string // ID сделки bj
	LKInvestorID    string // investor.id
	ContractNumber  string // № договора клиента
	SignedDocument  string // подпись клиента (base64)
	SignatureFormat string // формат подписи
	RequestXML      string // M2MTransferRequest (UTF-8, аудит)
}

// lkInsertRequest — заявка модуля ЛК (шапка и бумаги) в lk.transfer_requests / lk.transfer_securities.
func lkInsertRequest(ctx context.Context, rc *RuntimeConfig, req *m2m.M2MTransferRequest, meta LKRequestMeta) (int64, error) {
	db, err := lkDB(rc)
	if err != nil {
		return 0, err
	}
	cctx, cancel := context.WithTimeout(ctx, 30*time.Second)
	defer cancel()
	costInfo, costCode := "No", ""
	if req.Header.CostInfo.Yes != nil {
		costInfo, costCode = "Yes", string(req.Header.CostInfo.Yes.Code)
	}
	inv := req.Data.InvestorInformation
	doc := inv.IdentityDocument
	docSeries := ""
	if doc.DocumentSeries != nil {
		docSeries = string(*doc.DocumentSeries)
	}
	ourINN := ""
	if globalRC != nil {
		ourINN = transferringINN(globalRC.Snapshot())
	}
	opType := ""
	switch ourINN {
	case string(req.Data.TransferringDepository.INN):
		opType = "debit"
	case string(req.Data.ReceivingDepository.INN):
		opType = "credit"
	}
	tx, err := db.BeginTx(cctx, nil)
	if err != nil {
		return 0, err
	}
	var id int64
	err = tx.QueryRowContext(cctx, `INSERT INTO lk.transfer_requests (source, bj_claim_id, lk_investor_id, signed_document, signature_format,
		guid, creation_timestamp, sender_code, receiver_code, cost_info, cost_code, investor_last, investor_first, investor_middle,
		doc_type, doc_series, doc_number, transferring_inn, receiving_inn, operation_type, contract_number, request_xml, status)
		VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18,$19,$20,$21,$22,'new') RETURNING id`,
		meta.Source, meta.BJClaimID, meta.LKInvestorID, meta.SignedDocument, meta.SignatureFormat,
		string(req.Header.GUID), req.Header.CreationTimestamp.String(), string(req.Header.SenderCode), string(req.Header.ReceiverCode), costInfo, costCode,
		inv.LastName, inv.FirstName, inv.MiddleName, doc.DocumentType, docSeries, string(doc.DocumentNumber),
		string(req.Data.TransferringDepository.INN), string(req.Data.ReceivingDepository.INN), opType, meta.ContractNumber, meta.RequestXML).Scan(&id)
	if err != nil {
		tx.Rollback()
		return 0, fmt.Errorf("запись заявки: %w", err)
	}
	for i, sec := range req.Data.TransferredSecurities.Securities {
		isin := ""
		if sec.SecurityDetails.ISIN != nil {
			isin = string(*sec.SecurityDetails.ISIN)
		}
		var qWhole, qFrac any
		if sec.Quantity.Whole != nil {
			qWhole = int64(*sec.Quantity.Whole)
		} else if sec.Quantity.Fractional != nil {
			qFrac = string(*sec.Quantity.Fractional)
		}
		settINN, depCode, accID, sectID := "", "", "", ""
		if len(sec.SettlementAccount) > 0 {
			sa := sec.SettlementAccount[0]
			settINN, depCode, accID, sectID = string(sa.SettlementRequisites.INN), sa.SettlementLocation.DeponentCode, string(sa.SettlementLocation.AccountID), sa.SettlementLocation.SectionID
		}
		settJSON := ""
		if len(sec.SettlementAccount) > 1 {
			if b, e := json.Marshal(sec.SettlementAccount); e == nil {
				settJSON = string(b)
			}
		}
		if _, err := tx.ExecContext(cctx, `INSERT INTO lk.transfer_securities (request_id, seq, reference_id, security_code, isin,
			quantity_whole, quantity_fractional, settlement_inn, deponent_code, account_id, section_id, settlement_count, settlements_json, isolation_status)
			VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14)`,
			id, i+1, string(sec.ReferenceID), string(sec.SecurityCode), isin, qWhole, qFrac, settINN, depCode, accID, sectID,
			len(sec.SettlementAccount), settJSON, string(sec.IsolationStatus)); err != nil {
			tx.Rollback()
			return 0, fmt.Errorf("запись бумаги %d: %w", i+1, err)
		}
	}
	return id, tx.Commit()
}

// lkMarkStatus — статус заявки модуля ЛК (sent | error | ...).
func lkMarkStatus(ctx context.Context, rc *RuntimeConfig, id int64, status string) error {
	db, err := lkDB(rc)
	if err != nil {
		return err
	}
	cctx, cancel := context.WithTimeout(ctx, 12*time.Second)
	defer cancel()
	_, err = db.ExecContext(cctx, "UPDATE lk.transfer_requests SET status=$1 WHERE id=$2", status, id)
	return err
}

// lkFinalizeByGUID — итог НРД для заявки модуля ЛК: статус (done | declined) и причина,
// чтобы клиент видел результат. Заявки не из модуля ЛК молча пропускаются.
func lkFinalizeByGUID(ctx context.Context, rc *RuntimeConfig, guid, status, verdict, reason string) error {
	db, err := lkDB(rc)
	if err != nil {
		return nil // PostgreSQL не настроена — модуля ЛК нет
	}
	cctx, cancel := context.WithTimeout(ctx, 12*time.Second)
	defer cancel()
	var id int64
	if err := db.QueryRowContext(cctx, "SELECT id FROM lk.transfer_requests WHERE guid=$1", guid).Scan(&id); err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			return nil
		}
		return err
	}
	if _, err := db.ExecContext(cctx, "UPDATE lk.transfer_requests SET status=$1 WHERE id=$2", status, id); err != nil {
		return err
	}
	_, err = db.ExecContext(cctx, `INSERT INTO lk.transfer_responses (request_id, guid, verdict, reason, checked_by, check_source)
		VALUES ($1,$2,$3,$4,'НРД','MOEX МОСТ')`, id, guid, verdict, reason)
	return err
}

// lkClaimsByContract — заявки клиента модуля ЛК по номеру договора.
func lkClaimsByContract(ctx context.Context, rc *RuntimeConfig, contract string, limit int) ([]LKClaimRow, error) {
	db, err := lkDB(rc)
	if err != nil {
		return nil, err
	}
	cctx, cancel := context.WithTimeout(ctx, 15*time.Second)
	defer cancel()
	if limit <= 0 || limit > 500 {
		limit = 100
	}
	rows, err := db.QueryContext(cctx, `SELECT r.id, r.guid, r.status, to_char(r.created_at,'DD.MM.YYYY HH24:MI'),
		(SELECT COUNT(*) FROM lk.transfer_securities x WHERE x.request_id=r.id),
		COALESCE((SELECT reason FROM lk.transfer_responses y WHERE y.request_id=r.id ORDER BY y.id DESC LIMIT 1),'')
		FROM lk.transfer_requests r WHERE r.contract_number=$1 ORDER BY r.id DESC LIMIT $2`, contract, limit)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var out []LKClaimRow
	for rows.Next() {
		var c LKClaimRow
		if err := rows.Scan(&c.ID, &c.GUID, &c.Status, &c.CreatedAt, &c.Securit, &c.Reason); err != nil {
			return out, err
		}
		c.StatusRu = lkStatusRu(c.Status)
		out = append(out, c)
	}
	return out, rows.Err()
}