// contentsync 把源库(91porn_video)的帖子/视频/短剧增量同步到若干目标库。
//
// 增量策略（每个集合、每个目标库各自记水位线）：
//   - 新增：按 _id 升序游标，_id > lastID。ObjectId 单调递增且天然有索引。
//   - 修改：按 tsField > lastTS。仅对有该字段且建了索引的集合启用。
//     没有可靠时间字段的集合（media_content/media_tag/tag）靠定期 --full 兜底。
//
// 写入语义：$set 内容字段 + $setOnInsert 站点本地字段（见 spec.go）。
// 只更新不删除；源库下架通过 status 字段自然同步。
package main

import (
	"context"
	"encoding/json"
	"flag"
	"fmt"
	"os"
	"strings"
	"time"

	"go.mongodb.org/mongo-driver/bson"
	"go.mongodb.org/mongo-driver/bson/primitive"
	"go.mongodb.org/mongo-driver/mongo"
	"go.mongodb.org/mongo-driver/mongo/options"
)

// 源库名由 --src-db 指定。两个源库 91porn_video 与 91dj_video 老数据同源但已分叉
// （最新 500 条零重叠），所以每个目标库只能配一个源，不能混。
const defaultSrcDBName = "91porn_video"

type mark struct {
	LastID string    `json:"lastId"`
	LastTS time.Time `json:"lastTs"`
}

// state: target -> collection -> mark
type state map[string]map[string]*mark

func loadState(p string) state {
	s := state{}
	b, err := os.ReadFile(p)
	if err != nil {
		return s
	}
	_ = json.Unmarshal(b, &s)
	return s
}

func saveState(p string, s state) error {
	b, err := json.MarshalIndent(s, "", "  ")
	if err != nil {
		return err
	}
	tmp := p + ".tmp"
	if err := os.WriteFile(tmp, b, 0o600); err != nil {
		return err
	}
	return os.Rename(tmp, p) // 原子替换，避免崩溃时留下半截状态
}

func (s state) get(target, coll string) *mark {
	if s[target] == nil {
		s[target] = map[string]*mark{}
	}
	if s[target][coll] == nil {
		s[target][coll] = &mark{}
	}
	return s[target][coll]
}

// dstList 让 --dst 可以重复出现。MongoDB URI 自身含逗号（副本集主机列表），
// 所以不能用逗号分隔多个目标。
type dstList []string

func (d *dstList) String() string { return strings.Join(*d, " ") }
func (d *dstList) Set(v string) error {
	*d = append(*d, v)
	return nil
}

type target struct {
	name string
	db   *mongo.Database
}

func main() {
	var (
		srcURI    = flag.String("src", "", "源库 URI")
		dsts      dstList
		srcDBName = flag.String("src-db", defaultSrcDBName, "源库名")
		statePath = flag.String("state", "/server/sync/state.json", "水位线状态文件")
		full      = flag.Bool("full", false, "忽略水位线做全量扫描（用于兜底修复）")
		dryRun    = flag.Bool("dry-run", false, "只统计不写入")
		only      = flag.String("only", "", "只同步指定集合，逗号分隔")
		batchSize = flag.Int("batch", 500, "每批写入条数")
		sweep     = flag.Bool("sweep-deleted", true, "回传源库软删除标记（清理重复剧集）")
		timeout   = flag.Duration("timeout", 30*time.Minute, "整体超时")
	)
	flag.Var(&dsts, "dst", "目标库，格式 name=uri；可重复指定多个")
	flag.Parse()

	if *srcURI == "" || len(dsts) == 0 {
		fmt.Fprintln(os.Stderr, "必须指定 --src 和 --dst")
		os.Exit(2)
	}

	ctx, cancel := context.WithTimeout(context.Background(), *timeout)
	defer cancel()

	srcCli, err := mongo.Connect(ctx, options.Client().ApplyURI(*srcURI))
	if err != nil {
		fatal("连接源库失败: %v", err)
	}
	defer srcCli.Disconnect(context.Background())
	srcDB := srcCli.Database(*srcDBName)

	var targets []target
	for _, part := range dsts {
		part = strings.TrimSpace(part)
		if part == "" {
			continue
		}
		name, uri, ok := strings.Cut(part, "=")
		if !ok {
			fatal("--dst 格式错误，应为 name=uri: %s", part)
		}
		cli, err := mongo.Connect(ctx, options.Client().ApplyURI(uri))
		if err != nil {
			fatal("连接目标库 %s 失败: %v", name, err)
		}
		defer cli.Disconnect(context.Background())
		targets = append(targets, target{name: name, db: cli.Database(name)})
	}

	wanted := map[string]bool{}
	for _, n := range strings.Split(*only, ",") {
		if n = strings.TrimSpace(n); n != "" {
			wanted[n] = true
		}
	}

	st := loadState(*statePath)
	started := time.Now()
	var totalUp, totalIns int
	failed := false

	for _, spec := range specs {
		if len(wanted) > 0 && !wanted[spec.Name] {
			continue
		}
		for _, t := range targets {
			ins, up, err := syncOne(ctx, srcDB, t, spec, st, *full, *dryRun, *batchSize)
			if err != nil {
				fmt.Printf("[FAIL] %-20s -> %-10s %v\n", spec.Name, t.name, err)
				failed = true
				continue
			}
			totalIns += ins
			totalUp += up
			fmt.Printf("[ok]   %-20s -> %-10s 新增 %6d  更新 %6d\n", spec.Name, t.name, ins, up)
			if !*dryRun {
				if err := saveState(*statePath, st); err != nil {
					fmt.Printf("[WARN] 写状态文件失败: %v\n", err)
				}
			}
		}
	}

	if *sweep {
		for _, coll := range tombstoneColls {
			if len(wanted) > 0 && !wanted[coll] {
				continue
			}
			for _, t := range targets {
				n, fixed, err := sweepDeleted(ctx, srcDB, t, coll, *dryRun)
				if err != nil {
					fmt.Printf("[FAIL] 补扫软删除 %-14s -> %-10s %v\n", coll, t.name, err)
					failed = true
					continue
				}
				fmt.Printf("[ok]   补扫软删除 %-14s -> %-10s 标记 %5d  重算父级 %4d\n", coll, t.name, n, fixed)
			}
		}
	}

	fmt.Printf("\n合计 新增 %d 更新 %d，耗时 %s\n", totalIns, totalUp, time.Since(started).Round(time.Second))
	if failed {
		os.Exit(1)
	}
}

func syncOne(ctx context.Context, srcDB *mongo.Database, t target, spec CollSpec,
	st state, full, dryRun bool, batchSize int) (ins, up int, err error) {

	m := st.get(t.name, spec.Name)
	local := map[string]bool{}
	for _, f := range spec.LocalFields {
		local[f] = true
	}

	// 只按 _id 水位线取新增。ObjectId 单调递增且天然有索引，
	// 所以每次只读上次同步之后产生的文档，跟集合总量无关。
	//
	// 刻意不做存量修改比对：源库改了已存在的文档不会同步过来。
	// 需要补齐时用 --full 手动重扫一次。
	filter := bson.M{}
	if !full && m.LastID != "" {
		oid, e := primitive.ObjectIDFromHex(m.LastID)
		if e == nil {
			filter["_id"] = bson.M{"$gt": oid}
		}
	}
	ins, up, maxID, err := runPass(ctx, srcDB, t, spec, local, filter,
		bson.D{{Key: "_id", Value: 1}}, dryRun, batchSize)
	if err != nil {
		return ins, up, err
	}
	// 水位线只在真跑时推进，干跑不能改状态
	if !dryRun && maxID != primitive.NilObjectID {
		m.LastID = maxID.Hex()
	}
	return ins, up, nil
}

func runPass(ctx context.Context, srcDB *mongo.Database, t target, spec CollSpec,
	local map[string]bool, filter bson.M, sortBy bson.D, dryRun bool, batchSize int,
) (ins, up int, maxID primitive.ObjectID, err error) {

	srcCur, err := srcDB.Collection(spec.Name).Find(ctx, filter,
		options.Find().SetSort(sortBy).SetBatchSize(int32(batchSize)).SetNoCursorTimeout(true))
	if err != nil {
		return 0, 0, maxID, err
	}
	defer srcCur.Close(ctx)

	dst := t.db.Collection(spec.Name)
	pending := make([]bson.M, 0, batchSize)
	scanned, skipped, deduped := 0, 0, 0
	lastLog := time.Now()

	// blockedIDs 查出这批里哪些在目标库已被本地下架(status:0)。
	// 目标库没有 status 字段的集合自然查不出东西，等于不生效。
	blockedIDs := func(docs []bson.M) (map[any]bool, error) {
		out := map[any]bool{}
		ids := make([]any, 0, len(docs))
		for _, d := range docs {
			ids = append(ids, d["_id"])
		}
		c, e := dst.Find(ctx, bson.M{"_id": bson.M{"$in": ids}, "status": 0},
			options.Find().SetProjection(bson.M{"_id": 1}))
		if e != nil {
			return nil, e
		}
		defer c.Close(ctx)
		for c.Next(ctx) {
			var row bson.M
			if e := c.Decode(&row); e == nil {
				out[row["_id"]] = true
			}
		}
		return out, c.Err()
	}

	// localTwins 查出这批源文档在目标库的「同一集另一条」：同属一个 mediaId，
	// 且 hashId 或 videoUrl 相同，但 _id 不同。
	//
	// 各站点自己也在跑媒资库同步(common/media/syncMedia.go)，它按
	// (mediaId, hashId) 去重、用自己的 _id 为同一集插一条；本同步再把源库那条
	// 插进来，同一集就变成两条，前端剧集列表显示「1 1 2 2」。
	// 更要紧的是媒资库那条在 SaleCoin=0 时 listenPermission=2（免费），
	// 等于用户可以白看本该收费的集。
	//
	// 必须同时认 videoUrl：媒资库会给同一集重新发 hashId（实测
	// 「JK学姐不为人知的秘密」源库那条 hashId 6aa23652…、本地那条
	// 6acaf74c…，videoUrl 却一模一样），只比 hashId 会漏掉这一类。
	//
	// 以源库那条为准（购买记录都挂在它上面），把本地那条下架 isActive:false，
	// 不删除，便于事后回查。
	localTwins := func(docs []bson.M) (map[any]any, error) {
		keys := make([]any, 0, len(docs)*2) // 待查的 hashId 与 videoUrl
		want := map[string]any{}            // mediaId|key -> 源库 _id
		add := func(mid primitive.ObjectID, v string, id any) {
			if v == "" {
				return
			}
			keys = append(keys, v)
			want[mid.Hex()+"|"+v] = id
		}
		for _, d := range docs {
			mid, ok := d["mediaId"].(primitive.ObjectID)
			if !ok {
				continue
			}
			h, _ := d["hashId"].(string)
			u, _ := d["videoUrl"].(string)
			add(mid, h, d["_id"])
			add(mid, u, d["_id"])
		}
		if len(keys) == 0 {
			return nil, nil
		}
		// hashId 上有索引；videoUrl 没有，所以放在同一个 $or 里一次查完
		c, e := dst.Find(ctx, bson.M{"isActive": true, "$or": []bson.M{
			{"hashId": bson.M{"$in": keys}},
			{"videoUrl": bson.M{"$in": keys}},
		}}, options.Find().SetProjection(bson.M{"_id": 1, "mediaId": 1, "hashId": 1, "videoUrl": 1}))
		if e != nil {
			return nil, e
		}
		defer c.Close(ctx)
		out := map[any]any{}
		for c.Next(ctx) {
			var row struct {
				ID       primitive.ObjectID `bson:"_id"`
				MediaID  primitive.ObjectID `bson:"mediaId"`
				HashID   string             `bson:"hashId"`
				VideoURL string             `bson:"videoUrl"`
			}
			if e := c.Decode(&row); e != nil {
				return nil, e
			}
			for _, k := range []string{row.HashID, row.VideoURL} {
				if k == "" {
					continue
				}
				srcID, ok := want[row.MediaID.Hex()+"|"+k]
				if !ok {
					continue
				}
				if sid, ok := srcID.(primitive.ObjectID); ok && sid == row.ID {
					continue // 就是源库那条自己，不是重复
				}
				out[srcID] = row.ID
				break
			}
		}
		return out, c.Err()
	}

	flush := func() error {
		if len(pending) == 0 {
			return nil
		}
		docs := pending
		pending = make([]bson.M, 0, batchSize)

		blocked, e := blockedIDs(docs)
		if e != nil {
			return e
		}
		twins, e := localTwins(docs)
		if e != nil {
			return e
		}

		models := make([]mongo.WriteModel, 0, len(docs))
		for _, doc := range docs {
			id := doc["_id"]
			// 站点运营下架之后，源库无权再改它——整条跳过，内容字段也不写。
			if blocked[id] {
				skipped++
				continue
			}
			set := bson.M{}
			onInsert := bson.M{}
			for k, v := range doc {
				if k == "_id" {
					continue
				}
				if local[k] {
					onInsert[k] = v
				} else {
					set[k] = v
				}
			}
			update := bson.M{}
			if len(set) > 0 {
				update["$set"] = set
			}
			if len(onInsert) > 0 {
				update["$setOnInsert"] = onInsert
			}
			if len(update) == 0 {
				continue
			}
			models = append(models, mongo.NewUpdateOneModel().
				SetFilter(bson.M{"_id": id}).SetUpdate(update).SetUpsert(true))
			// 同一集本地已有另一条（媒资库同步插的），把它下架
			if twinID, ok := twins[id]; ok {
				models = append(models, mongo.NewUpdateOneModel().
					SetFilter(bson.M{"_id": twinID}).
					SetUpdate(bson.M{"$set": bson.M{"isActive": false}}))
				deduped++
			}
		}
		if len(models) == 0 {
			return nil
		}
		if dryRun {
			up += len(models)
			return nil
		}
		res, e := dst.BulkWrite(ctx, models, options.BulkWrite().SetOrdered(false))
		if e != nil {
			if res != nil {
				ins += int(res.UpsertedCount)
				up += int(res.ModifiedCount)
			}
			return e
		}
		ins += int(res.UpsertedCount)
		up += int(res.ModifiedCount)
		return nil
	}

	for srcCur.Next(ctx) {
		var doc bson.M
		if e := srcCur.Decode(&doc); e != nil {
			return ins, up, maxID, e
		}
		scanned++
		// 大集合动辄几十万条，cron 里长时间无输出无法判断死活
		if time.Since(lastLog) >= 30*time.Second {
			fmt.Printf("       %-20s -> %-10s 已扫描 %d 条…\n", spec.Name, t.name, scanned)
			lastLog = time.Now()
		}
		// 水位线推进到扫过的最大 _id。即使这条因本地下架被跳过也要推进，
		// 否则它会永久卡住水位线，导致每次都从它开始重扫。
		if id, ok := doc["_id"].(primitive.ObjectID); ok && !id.IsZero() {
			if maxID.IsZero() || id.Hex() > maxID.Hex() {
				maxID = id
			}
		}
		pending = append(pending, doc)
		if len(pending) >= batchSize {
			if e := flush(); e != nil {
				return ins, up, maxID, e
			}
		}
	}
	if e := srcCur.Err(); e != nil {
		return ins, up, maxID, e
	}
	if e := flush(); e != nil {
		return ins, up, maxID, e
	}
	if skipped > 0 {
		fmt.Printf("       %-20s -> %-10s 跳过 %d 条（本地已下架）\n", spec.Name, t.name, skipped)
	}
	if deduped > 0 {
		fmt.Printf("       %-20s -> %-10s 下架 %d 条本地重复集（媒资库同步插的同一集）\n", spec.Name, t.name, deduped)
	}
	return ins, up, maxID, nil
}

// tombstoneColls 需要回传源库软删除标记的集合。
//
// 常规同步只按 _id 水位线插新增，源库事后打上的 isDelete 永远传不过来：
// 源站把重复剧集软删掉了，目标库却还留着，App 的剧集列表就会出现
// 「1 1 1 2 2 2」这种重复集。每轮补扫一次把标记补齐。
//
// 只做「源库删了 → 目标库也删」这一个方向，绝不把目标库已删的恢复回来，
// 所以和「本地下架后不再同步」不冲突。
var tombstoneColls = []string{"media", "media_content", "video_info"}

// sweepDeleted 把源库 isDelete:true 的文档在目标库也标成已删除，
// 并按 web/service/mediacontentser 的同一口径重算父级媒体的集数。
func sweepDeleted(ctx context.Context, srcDB *mongo.Database, t target, coll string, dryRun bool,
) (marked, fixed int, err error) {

	cur, err := srcDB.Collection(coll).Find(ctx, bson.M{"isDelete": true},
		options.Find().SetProjection(bson.M{"_id": 1}).SetNoCursorTimeout(true))
	if err != nil {
		return 0, 0, err
	}
	defer cur.Close(ctx)

	dst := t.db.Collection(coll)
	parents := map[primitive.ObjectID]bool{}
	batch := make([]primitive.ObjectID, 0, 1000)

	flush := func() error {
		if len(batch) == 0 {
			return nil
		}
		ids := batch
		batch = make([]primitive.ObjectID, 0, 1000)

		// 先把本轮真正要改的那几条捞出来，顺带记下父级，用于之后重算集数
		hit := make([]primitive.ObjectID, 0, len(ids))
		c, e := dst.Find(ctx, bson.M{"_id": bson.M{"$in": ids}, "isDelete": false},
			options.Find().SetProjection(bson.M{"_id": 1, "mediaId": 1}))
		if e != nil {
			return e
		}
		defer c.Close(ctx)
		for c.Next(ctx) {
			var row struct {
				ID      primitive.ObjectID `bson:"_id"`
				MediaID primitive.ObjectID `bson:"mediaId"`
			}
			if e := c.Decode(&row); e != nil {
				return e
			}
			hit = append(hit, row.ID)
			if coll == "media_content" && !row.MediaID.IsZero() {
				parents[row.MediaID] = true
			} else if coll == "media" {
				parents[row.ID] = true
			}
		}
		if e := c.Err(); e != nil {
			return e
		}
		if len(hit) == 0 {
			return nil
		}
		if dryRun {
			marked += len(hit)
			return nil
		}
		// 只写 isDelete，不碰 updateTime —— 那是站点本地字段
		res, e := dst.UpdateMany(ctx, bson.M{"_id": bson.M{"$in": hit}},
			bson.M{"$set": bson.M{"isDelete": true}})
		if e != nil {
			return e
		}
		marked += int(res.ModifiedCount)
		return nil
	}

	for cur.Next(ctx) {
		var row struct {
			ID primitive.ObjectID `bson:"_id"`
		}
		if e := cur.Decode(&row); e != nil {
			return marked, fixed, e
		}
		batch = append(batch, row.ID)
		if len(batch) >= 1000 {
			if e := flush(); e != nil {
				return marked, fixed, e
			}
		}
	}
	if e := cur.Err(); e != nil {
		return marked, fixed, e
	}
	if e := flush(); e != nil {
		return marked, fixed, e
	}

	if dryRun || len(parents) == 0 {
		return marked, len(parents), nil
	}

	// 父级的 currentEpisode/freeEpisode 原本只在后台编辑子集时重算，
	// 这里补上，否则集数会停留在含重复集的旧值（如 2 集的剧显示 6 集）。
	mc := t.db.Collection("media_content")
	media := t.db.Collection("media")
	for mid := range parents {
		base := bson.M{"mediaId": mid, "isActive": true, "isDelete": false}
		cnt, e := mc.CountDocuments(ctx, base)
		if e != nil {
			return marked, fixed, e
		}
		free, e := mc.CountDocuments(ctx, bson.M{"mediaId": mid, "listenPermission": 2,
			"isActive": true, "isDelete": false})
		if e != nil {
			return marked, fixed, e
		}
		res, e := media.UpdateOne(ctx, bson.M{"_id": mid},
			bson.M{"$set": bson.M{"currentEpisode": cnt, "freeEpisode": free}})
		if e != nil {
			return marked, fixed, e
		}
		fixed += int(res.ModifiedCount)
	}
	return marked, fixed, nil
}

func fatal(f string, a ...any) {
	fmt.Fprintf(os.Stderr, f+"\n", a...)
	os.Exit(1)
}