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

package lkgateway

import (
	"context"
	"encoding/json"
	"log"
	"os"
	"strings"
	"time"

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

// pollerStateFile — постоянная память id применённых входящих пакетов ИШ.
// Без неё перезапуск bj-server заново вычитывал бы старые ответы НРД и
// повторно применял их к заявкам (для ответов с нулевым GUID это приводило
// к ложным «отклонениям» по FIFO). Файл переживает перезапуски.
const pollerStateFile = "/var/lib/bj/.bj/poller-processed.json"

// loadProcessed читает множество обработанных id; второй результат false,
// если файла нет (самый первый запуск).
func loadProcessed() (map[int]bool, bool) {
	m := make(map[int]bool)
	b, err := os.ReadFile(pollerStateFile)
	if err != nil {
		return m, false
	}
	var ids []int
	if json.Unmarshal(b, &ids) != nil {
		return m, false
	}
	for _, id := range ids {
		m[id] = true
	}
	return m, true
}

// saveProcessed атомарно сохраняет множество обработанных id.
func saveProcessed(m map[int]bool) {
	ids := make([]int, 0, len(m))
	for id := range m {
		ids = append(ids, id)
	}
	b, err := json.Marshal(ids)
	if err != nil {
		return
	}
	tmp := pollerStateFile + ".tmp"
	if os.WriteFile(tmp, b, 0o640) == nil {
		_ = os.Rename(tmp, pollerStateFile)
	}
}

// pollIncoming периодически опрашивает ИШ на входящие пакеты от НРД
// (M2MTransferDecision / Response) и применяет их через svc.ApplyDecision.
// Замыкает цикл: bj-server отправил заявку → ИШ → НРД → робот ответил →
// ИШ забрал ответ во входящие → этот поллер применяет Decision (статус
// заявки переходит в confirmed/rejected, срабатывает callback в ЛК).
//
// Дедупликация по id обработанных пакетов: ИШ возвращает их повторно,
// пока мы не подтвердим, поэтому держим множество уже обработанных.
func (s *Server) pollIncoming(ctx context.Context) {
	const interval = 10 * time.Second // синхронно с fetchInterval ИШ (10с)
	ticker := time.NewTicker(interval)
	defer ticker.Stop()

	processed, existed := loadProcessed()
	if !existed {
		// Первый запуск: помечаем все уже лежащие во входящих пакеты как
		// обработанные, чтобы не применять исторический backlog (старые ответы
		// НРД) к текущим заявкам. Реальные новые ответы придут позже.
		s.seedProcessed(ctx, processed)
		saveProcessed(processed)
	}
	log.Printf("lk-gateway: поллер входящих ИШ запущен (канал %s, интервал %s, обработанных в памяти %d)", s.igwChannel, interval, len(processed))
	go s.participantReceiptBackfill(ctx)

	for {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
			healthBeat("poll-incoming")
			s.fetchAndApply(ctx, processed)
		}
	}
}

// fetchAndApply — один проход поллера: список входящих → для каждого нового
// забираем тело, распаковываем, парсим Decision, применяем.
func (s *Server) fetchAndApply(ctx context.Context, processed map[int]bool) {
	cctx, cancel := context.WithTimeout(ctx, 25*time.Second)
	defer cancel()

	// Тип не указываем — ИШ вернёт оба (M2MTD + M2MER). Date=сегодня.
	pkgs, err := s.igwClient.ListIncoming(cctx, igw.ListFilter{
		Channel: s.igwChannel,
		Date:    time.Now(),
	})
	if err != nil {
		log.Printf("lk-gateway: поллер ListIncoming: %v", err)
		return
	}

	for _, p := range pkgs {
		if processed[p.ID] {
			continue
		}
		if err := s.applyIncoming(cctx, p); err != nil {
			log.Printf("lk-gateway: поллер пакет id=%d (%s): %v", p.ID, p.Type, err)
			continue // не помечаем обработанным — повторим в след. раз
		}
		processed[p.ID] = true
		saveProcessed(processed) // переживает перезапуск
		log.Printf("lk-gateway: поллер применил входящий пакет id=%d тип=%s", p.ID, p.Type)
	}
}

// seedProcessed помечает все текущие входящие пакеты как обработанные
// (используется при самом первом запуске поллера).
func (s *Server) seedProcessed(ctx context.Context, processed map[int]bool) {
	cctx, cancel := context.WithTimeout(ctx, 25*time.Second)
	defer cancel()
	pkgs, err := s.igwClient.ListIncoming(cctx, igw.ListFilter{Channel: s.igwChannel, Date: time.Now()})
	if err != nil {
		log.Printf("lk-gateway: поллер seed: ListIncoming: %v", err)
		return
	}
	for _, p := range pkgs {
		processed[p.ID] = true
	}
	log.Printf("lk-gateway: поллер первый запуск — засеяно %d существующих пакетов как обработанные", len(pkgs))
}

// applyIncoming забирает тело пакета и применяет M2M-ответ к сделке.
// Среди входящих от НРД много служебных пакетов (квитанции ЭДО типа C/Z,
// конверты) — они не M2M-документы и пропускаются. Реальные ответы —
// M2MTransferDecision (решение принимающей стороны) и M2MTransferResponse
// (ответ сервиса МОСТ, в т.ч. ошибки M2Mxx).
func (s *Server) applyIncoming(ctx context.Context, p igw.Package) error {
	zipBytes, err := s.igwClient.GetPackage(ctx, p.ID)
	if err != nil {
		return err // сетевая ошибка — повторим в след. раз
	}
	unpacked, err := igw.UnpackPackage(zipBytes)
	if err != nil {
		// Нет основного .xml — служебный пакет (квитанция/конверт ЭДО).
		// Не ошибка: помечаем обработанным, чтобы не повторять. Квитанция о получении нашей анкеты —
		// отметка «доставлена в НРД» (nsdreceipt.go).
		if r, ok := parseNSDReceipt(zipBytes); ok {
			if applyNSDReceipt(r) {
				log.Printf("lk-gateway: поллер квитанция id=%d: анкета %s получена НРД %s, без ошибок=%v %s", p.ID, r.Package, r.ReceivedAt.Format("02.01 15:04:05"), r.OK, r.Text)
			}
		}
		log.Printf("lk-gateway: поллер пакет id=%d (%s) — служебный (квитанция/конверт), пропуск", p.ID, p.Type)
		return nil
	}
	doc := string(unpacked.DocXML)
	switch {
	case strings.Contains(doc, "M2MTransferDecision"):
		decision, err := igw.ParseDecision(unpacked.DocXML)
		if err != nil {
			log.Printf("lk-gateway: поллер Decision id=%d: разбор: %v", p.ID, err)
			return nil
		}
		s.applyDecisionEffects(ctx, decision, "поллер", "")
		return nil
	case strings.Contains(doc, "M2MTransferResponse"):
		resp, err := igw.ParseResponse(unpacked.DocXML)
		if err != nil {
			log.Printf("lk-gateway: поллер Response id=%d: разбор: %v", p.ID, err)
			return nil
		}
		// Ответ сервиса МОСТ: статус + код (M2Mxx). Применяем к сделке:
		// INFO — приём в обработку (статус не меняется), ERROR — отказ сервиса
		// (напр. M2M14 — отправитель не в справочнике), сделка → Отклонена.
		// Ответ сохраняется в сделке и виден в карточке заявки.
		var codes string
		for _, rr := range resp.Responses {
			codes += string(rr.Code) + " "
		}
		log.Printf("lk-gateway: поллер M2MTransferResponse id=%d: статус=%s коды=[%s] GUID=%s",
			p.ID, resp.StatusCode, strings.TrimSpace(codes), resp.GUID)
		if gs := string(resp.GUID); gs == "" || gs == zeroGUID {
			// Нулевой GUID — ответ не на перевод (сделки нет), а на регистрацию участника
			// (анкета #M2MPF) или служебное сообщение сервиса МОСТ. Фиксируем и показываем
			// на странице анкеты, а не отбрасываем как «сделка не найдена».
			raw, _ := captureNSDReply(p, unpacked.DocXML, "participant-form")
			recordParticipantReply(string(resp.StatusCode), resp, raw)
			log.Printf("lk-gateway: поллер ОТВЕТ НРД на анкету участника id=%d: %s [%s] %q (сохранён %s)",
				p.ID, resp.StatusCode, strings.TrimSpace(codes), strings.Join(serviceErrorTexts(resp), "; "), raw)
			return nil
		}
		if err := s.svc.ApplyServiceResponse(ctx, resp, unpacked.DocXML); err != nil {
			// Сделка может быть не найдена (ответ на чужой/старый GUID) —
			// логируем, но помечаем обработанным, чтобы не зациклиться.
			log.Printf("lk-gateway: поллер ApplyServiceResponse id=%d GUID=%s: %v", p.ID, resp.GUID, err)
		}
		// ERROR сервиса МОСТ — отказ: замыкаем на ЛК с человекочитаемой причиной.
		if strings.EqualFold(string(resp.StatusCode), "ERROR") {
			reason := "Отклонено сервисом MOEX МОСТ"
			if parts := serviceErrorTexts(resp); len(parts) > 0 {
				reason = strings.Join(parts, "; ")
			}
			// Итог для заявки модуля ЛК (lk.transfer_requests): 'declined' — отказ после отправки в НРД.
			if string(resp.GUID) != "" && string(resp.GUID) != zeroGUID {
				if e := lkFinalizeByGUID(ctx, s.rc, string(resp.GUID), "declined", "rejected", reason); e != nil {
					log.Printf("lk-gateway: поллер: итог заявки модуля ЛК GUID=%s: %v", resp.GUID, e)
				}
			}
			OnLKOutcome(ctx, s.rc, string(resp.GUID), "rejected", reason)
			quikOnServiceError(ctx, s.rc, string(resp.GUID), reason)
			go depoOnServiceError(s.rc, string(resp.GUID), codes, reason)
		}
		return nil
	case strings.Contains(doc, "M2MTransferRequest"):
		// Входящий запрос на перевод — другая сторона (робот в сценарии 3333)
		// выступает ПЕРЕДАЮЩЕЙ. Мы должны ответить M2MTransferDecision. Сохраняем
		// во «входящие переводы»; решение оператор формирует в UI.
		var req m2m.M2MTransferRequest
		// Робот шлёт корень <M2MTransferRequest> без namespace (префиксы только
		// на детях) — подставляем default namespace request для корня.
		if err := nsdxml.UnmarshalDefaultNS(unpacked.DocXML, &req, "http://nsd.ru/schemas/m2m/request"); err != nil {
			log.Printf("lk-gateway: поллер входящий Request id=%d: разбор: %v", p.ID, err)
			return nil
		}
		it, existed := s.incoming.AddFromRequest(&req, string(nsdxml.DecodeWindows1251(unpacked.DocXML)))
		if existed {
			log.Printf("lk-gateway: поллер входящий перевод GUID=%s уже сохранён (id=%s)", it.GUID, it.ID)
		} else {
			// Каждый новый GUID — ОТДЕЛЬНЫЙ перевод и требует решения оператора.
			// Совпадение по «отпечатку» (тот же инвестор и бумаги) НЕ значит дубль:
			// один и тот же человек может законно перевести 10 акций, затем ещё 10
			// (разные GUID = разные отправки). Поэтому авто-ответа нет — только
			// информируем, если запрос похож на ранее отвеченный.
			if dup, ok := s.incoming.FindAnsweredDuplicate(it.Fingerprint, it.ID); ok && it.Fingerprint != "" {
				log.Printf("lk-gateway: поллер входящий GUID=%s похож по составу на ранее отвеченный (id=%s) — "+
					"это отдельный перевод, ждёт решения", it.GUID, dup.ID)
			} else {
				log.Printf("lk-gateway: поллер НОВЫЙ входящий перевод от %s GUID=%s id=%s — ждёт решения",
					it.SenderCode, it.GUID, it.ID)
			}
			// Уведомляем депозитарий по почте (к вам переведены бумаги — учесть).
			go notifyDepositoryIncoming(s.rc, it)
			// DEPO (bj2depo): шапка + строки + BJ_RECEIVED — поручение в DEPO создаётся сразу.
			go depoOnIncomingReceived(s.rc, it)
			// Сверка с DEPO — только рекомендация для оператора (эмуляция ответа),
			// в статус попадает бейдж «сверка: принять / отказ M2Mxx». Не отвечаем.
			go func(item *IncomingTransfer) {
				if r := runDepoCheck(item); r.Ran {
					s.incoming.SetDepo(item.ID, r.Recommend, r.Summary)
					log.Printf("lk-gateway: сверка DEPO по входящему GUID=%s: %s (не отправлено)", item.GUID, r.Summary)
				} else {
					log.Printf("lk-gateway: сверка DEPO по входящему GUID=%s не выполнена: %s", item.GUID, r.Err)
				}
			}(it)
		}
		return nil
	case strings.Contains(doc, "M2MTransferHandbook") && !strings.Contains(doc, "M2MTransferHandbookRequest"):
		// Справочник участников M2M (ответ НРД на #M2MHR). Разбираем и сохраняем
		// — это карта ИНН → MC-код участника и перечень мест расчётов.
		var hb m2m.M2MTransferHandbook
		if err := nsdxml.Unmarshal(unpacked.DocXML, &hb); err != nil {
			log.Printf("lk-gateway: поллер справочник M2M id=%d: разбор: %v", p.ID, err)
			return nil
		}
		nsdProbeReply(time.Now())
		writeHandbookRawXML(unpacked.DocXML) // подлинный XML НРД — для выгрузки со страницы справочника
		changed := saveHandbook(&hb)
		participantHandbookCheck(s.rc) // появились в справочнике — анкета принята
		if !changed {
			log.Printf("lk-gateway: поллер справочник M2M id=%d: без изменений (%d участников) — связь с НРД подтверждена", p.ID, len(hb.Participants.Participants))
			return nil
		}
		go func() {
			if err := depoWriteHandbook(s.rc); err != nil {
				log.Printf("depo-events: справочник после получения: %v", err)
			}
		}()
		quikRecipientsSyncAsync(s.rc) // депозитарии-получатели в МНП (quikrecipients.go)
		log.Printf("lk-gateway: поллер СПРАВОЧНИК M2M получен id=%d: участников %d, мест расчётов %d (состав изменился)",
			p.ID, len(hb.Participants.Participants), len(hb.SettlementPlaces.Places))
		return nil
	case strings.Contains(doc, "M2MTransferParticipantForm"):
		// Ответ НРД на нашу анкету участника (#M2MPF): подтверждение регистрации.
		// Схему ответа фиксируем на боевом контуре — сохраняем подлинный XML, чтобы
		// затем встроить разбор в систему (для всех клиентов).
		path, root := captureNSDReply(p, unpacked.DocXML, "participant-form")
		recordParticipantFormReply(path)
		log.Printf("lk-gateway: поллер ОТВЕТ НРД на анкету участника id=%d (%s): корень <%s>, сохранён %s", p.ID, p.Type, root, path)
		return nil
	default:
		// Неизвестный документ НРД — тоже «отлавливаем»: сохраняем подлинный XML для разбора.
		path, root := captureNSDReply(p, unpacked.DocXML, "unknown")
		log.Printf("lk-gateway: поллер пакет id=%d (%s) — неизвестный M2M-документ <%s>, сохранён %s", p.ID, p.Type, root, path)
		return nil
	}
}

// rejectCodesLabel собирает человекочитаемые причины отказа по кодам M2M.
func rejectCodesLabel(codes []string) string {
	if len(codes) == 0 {
		return "без кода"
	}
	seen := map[string]bool{}
	var parts []string
	for _, c := range codes {
		if c == "" || seen[c] {
			continue
		}
		seen[c] = true
		parts = append(parts, m2mCodeLabel(c))
	}
	return strings.Join(parts, "; ")
}

// handbookFile — последний полученный справочник участников M2M (для UI/поиска
// ИНН→MC-код контрагента).
const handbookFile = "/var/lib/bj/.bj/m2m-handbook.json"

// HandbookEntry — плоская запись справочника для сохранения/показа.
type HandbookEntry struct {
	INN         string `json:"inn"`
	Code        string `json:"code"` // MC-код участника (депозитарий/брокер)
	Kind        string `json:"kind"` // "depo" | "broker" | "place"
	Name        string `json:"name"` // краткое (рус.)
	FullName    string `json:"full_name,omitempty"`
	DisplayName string `json:"display_name,omitempty"`
	EngName     string `json:"eng_name,omitempty"`
}

// saveHandbook раскладывает справочник в плоский список и пишет в файл. Возвращает, изменился ли
// состав: справочник запрашивается раз в час (сквозная проверка связи, nsdhealth.go), и без
// изменений его не нужно ни рассылать, ни переписывать в DEPO.
func saveHandbook(hb *m2m.M2MTransferHandbook) (changed bool) { return saveHandbookEx(hb, true) }

// saveHandbookEx — как saveHandbook, но push управляет рассылкой справочника на сервер
// лицензий. Ручная загрузка файла (handbook_import.go) идёт с push=false: справочник
// одного стенда не должен становиться облачным для всех клиентов.
func saveHandbookEx(hb *m2m.M2MTransferHandbook, push bool) (changed bool) {
	out := struct {
		At           string          `json:"at"`
		Participants []HandbookEntry `json:"participants"`
		Places       []HandbookEntry `json:"settlement_places"`
	}{At: time.Now().Format(time.RFC3339)}
	for _, p := range hb.Participants.Participants {
		e := HandbookEntry{INN: string(p.INN), Name: p.Names.Rus.ShortName, FullName: p.Names.Rus.FullName, DisplayName: p.Names.Rus.DisplayName}
		if p.Names.Eng != nil {
			e.EngName = p.Names.Eng.DisplayName
			if e.EngName == "" {
				e.EngName = p.Names.Eng.FullName
			}
		}
		if e.Name == "" {
			e.Name = e.DisplayName
		}
		if p.DepositoryPlace != nil {
			e.Code, e.Kind = string(p.DepositoryPlace.ParticipantCode), "depo"
		} else if p.BrokerPlace != nil {
			e.Code, e.Kind = string(p.BrokerPlace.ParticipantCode), "broker"
		}
		out.Participants = append(out.Participants, e)
	}
	for _, pl := range hb.SettlementPlaces.Places {
		out.Places = append(out.Places, HandbookEntry{INN: string(pl.INN), Name: pl.ShortName, DisplayName: pl.DisplayName, Kind: "place"})
	}
	changed = true
	if old, err := os.ReadFile(handbookFile); err == nil {
		var prev struct {
			Participants []HandbookEntry `json:"participants"`
			Places       []HandbookEntry `json:"settlement_places"`
		}
		if json.Unmarshal(old, &prev) == nil {
			a, _ := json.Marshal([]any{prev.Participants, prev.Places})
			b, _ := json.Marshal([]any{out.Participants, out.Places})
			changed = string(a) != string(b)
		}
	}
	if raw, err := json.Marshal(out); err == nil {
		_ = os.WriteFile(handbookFile, raw, 0o600)
		if changed {
			for _, e := range out.Participants {
				log.Printf("lk-gateway: справочник M2M: ИНН %s → %s [%s] %s", e.INN, e.Code, e.Kind, e.Name)
			}
			if push {
				pushHandbookAsync(globalRC, raw) // свежий справочник — на сервер лицензий для всех клиентов
			}
		}
	}
	return changed
}

// serviceErrorTexts — причины отказа сервиса M2M понятными словами: наша расшифровка кода, иначе текст
// из ответа НРД (он есть в каждом Response), иначе сам код. Раньше неизвестный bj код уходил клиенту
// в ЛК голым («M2M43»).
func serviceErrorTexts(resp *m2m.M2MTransferResponse) []string {
	var parts []string
	seen := map[string]bool{}
	for _, rr := range resp.Responses {
		code := strings.TrimSpace(rr.Code)
		if code == "" || seen[code] {
			continue
		}
		seen[code] = true
		switch {
		case m2mCodeText(code) != "":
			parts = append(parts, m2mCodeLabel(code))
		case strings.TrimSpace(rr.Text) != "":
			parts = append(parts, strings.TrimSpace(rr.Text)+" ("+code+")")
		default:
			parts = append(parts, code)
		}
	}
	return parts
}