ai_site/platform/internal/dbsync/reconcile.go

package dbsync

import (
	"context"
	"database/sql"
	"encoding/json"
	"fmt"
	"time"
)

type ReconcileReport struct {
	Table       string   `json:"table"`
	OnlyLocal   []string `json:"only_local"`   // 在 B 有、A 无 → 需推到线上
	OnlyRemote  []string `json:"only_remote"`  // 在 A 有、B 无 → 需拉到本地
	PatchedPush int      `json:"patched_push"`
	PatchedPull int      `json:"patched_pull"`
}

type ReconcileResult struct {
	ChannelID string           `json:"channel_id"`
	Reports   []ReconcileReport `json:"reports"`
	Message   string           `json:"message"`
}

// ListPKs 列出表全部主键（用于对账，防漏）。
func ListPKs(ctx context.Context, db *sql.DB, driver Driver, table, pkCol string) ([]string, error) {
	q := fmt.Sprintf(`SELECT %s FROM %s`, quoteIdent(driver, pkCol), quoteIdent(driver, table))
	rows, err := db.QueryContext(ctx, q)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var out []string
	for rows.Next() {
		var v any
		if err := rows.Scan(&v); err != nil {
			return nil, err
		}
		out = append(out, fmt.Sprint(v))
	}
	return out, rows.Err()
}

// ReconcileChannel 全量主键对账：只在一端的行补投递，保证不漏；已有行靠 outbox/版本，不重复插入。
func ReconcileChannel(ctx context.Context, ch *Channel) (*ReconcileResult, error) {
	local, err := Open(ch.Local.Driver, ch.Local.DSN)
	if err != nil {
		return nil, fmt.Errorf("open local: %w", err)
	}
	defer local.Close()
	remote, err := Open(ch.Remote.Driver, ch.Remote.DSN)
	if err != nil {
		return nil, fmt.Errorf("open remote: %w", err)
	}
	defer remote.Close()

	if err := prepareEndpoint(ctx, local, ch.Local, ch.PKColumns); err != nil {
		return nil, err
	}
	if err := prepareEndpoint(ctx, remote, ch.Remote, ch.PKColumns); err != nil {
		return nil, err
	}

	tables := uniqueTables(ch.Local.Tables, ch.Remote.Tables)
	res := &ReconcileResult{ChannelID: ch.ID, Message: "ok"}
	for _, table := range tables {
		pkCol := "id"
		if ch.PKColumns != nil && ch.PKColumns[table] != "" {
			pkCol = ch.PKColumns[table]
		}
		lpks, err := ListPKs(ctx, local, ch.Local.Driver, table, pkCol)
		if err != nil {
			return nil, fmt.Errorf("list local %s: %w", table, err)
		}
		rpks, err := ListPKs(ctx, remote, ch.Remote.Driver, table, pkCol)
		if err != nil {
			return nil, fmt.Errorf("list remote %s: %w", table, err)
		}
		lset := toSet(lpks)
		rset := toSet(rpks)
		rep := ReconcileReport{Table: table}
		for pk := range lset {
			if !rset[pk] {
				rep.OnlyLocal = append(rep.OnlyLocal, pk)
			}
		}
		for pk := range rset {
			if !lset[pk] {
				rep.OnlyRemote = append(rep.OnlyRemote, pk)
			}
		}
		// 补漏：缺的一端从有的一端取行并 apply（带 WithApplying，不产生回声）
		for _, pk := range rep.OnlyLocal {
			js, _, ferr := FetchRowJSON(ctx, local, ch.Local.Driver, table, pkCol, pk)
			if ferr != nil {
				continue
			}
			ver := time.Now().UnixMilli()
			if err := ApplyChange(ctx, remote, ch.Remote.Driver, table, pkCol, "upsert", js, ver); err == nil {
				rep.PatchedPush++
				_ = UpsertMeta(ctx, local, ch.Local.Driver, table, pk, ver)
			}
		}
		for _, pk := range rep.OnlyRemote {
			js, _, ferr := FetchRowJSON(ctx, remote, ch.Remote.Driver, table, pkCol, pk)
			if ferr != nil {
				continue
			}
			ver := time.Now().UnixMilli()
			if err := ApplyChange(ctx, local, ch.Local.Driver, table, pkCol, "upsert", js, ver); err == nil {
				rep.PatchedPull++
				_ = UpsertMeta(ctx, remote, ch.Remote.Driver, table, pk, ver)
			}
		}
		res.Reports = append(res.Reports, rep)
	}
	return res, nil
}

func uniqueTables(a, b []string) []string {
	m := map[string]struct{}{}
	var out []string
	for _, t := range append(append([]string{}, a...), b...) {
		if t == "" {
			continue
		}
		if _, ok := m[t]; ok {
			continue
		}
		m[t] = struct{}{}
		out = append(out, t)
	}
	return out
}

func toSet(xs []string) map[string]bool {
	m := make(map[string]bool, len(xs))
	for _, x := range xs {
		m[x] = true
	}
	return m
}

// IngestRows：外部源 C → 写入本地 B（会触发 outbox，再同步到线上 A）。
// 按主键 upsert，同一 PK 重复灌入不会「多」出一行。
func IngestRows(ctx context.Context, ch *Channel, table string, rows []map[string]any, sourceLabel string) (int, error) {
	if table == "" {
		return 0, fmt.Errorf("table required")
	}
	pkCol := "id"
	if ch.PKColumns != nil && ch.PKColumns[table] != "" {
		pkCol = ch.PKColumns[table]
	}
	db, err := Open(ch.Local.Driver, ch.Local.DSN)
	if err != nil {
		return 0, err
	}
	defer db.Close()
	if err := prepareEndpoint(ctx, db, ch.Local, ch.PKColumns); err != nil {
		return 0, err
	}
	n := 0
	for _, row := range rows {
		if row == nil {
			continue
		}
		pkVal := fmt.Sprint(row[pkCol])
		if pkVal == "" || pkVal == "<nil>" {
			return n, fmt.Errorf("row missing pk %s", pkCol)
		}
		// 不置 applying：让触发器写 outbox，随后 worker 推到 A
		b, _ := json.Marshal(row)
		ver := time.Now().UnixMilli()
		// 正常写入本地：触发器入 outbox → worker 推到线上 A；同 PK upsert 不会多行
		if err := applyChangeInner(ctx, db, ch.Local.Driver, table, pkCol, "upsert", string(b), ver); err != nil {
			return n, err
		}
		_ = sourceLabel
		n++
	}
	return n, nil
}