/** 标准领域事件：整条链路唯一在流转的对象 */
@Data
@Builder
public class BinlogEvent {
    private String table;         // 表名
    private OpType opType;        // INSERT / UPDATE / DELETE
    private long ts;              // 变更时间戳
    private String primaryKey;    // 主键值（用于幂等/去重）
    private JSONObject before;    // 变更前
    private JSONObject after;     // 变更后
}

/** 归一化器：每种外部格式一个实现，只做「识别 + 转成领域事件」 */
public interface FormatNormalizer {
    boolean support(String rawJson);
    BinlogEvent normalize(String rawJson);
}

@Component
public class BinlogConsumer {

    @Resource
    private BinlogPipeline pipeline;

    /** 批量入口：单条失败不影响其它消息 */
    public void onMessages(List<String> messages) {
        for (String raw : messages) {
            try {
                pipeline.accept(raw);
            } catch (Exception e) {
                // 交由 Pipeline 内部已分级处理，这里只兜最外层
                log.error("binlog 消费异常, raw={}", raw, e);
            }
        }
    }
}

@Component
public class BinlogPipeline {

    /** 1. 归一化器集合：屏蔽外部格式差异 */
    @Resource
    private List<FormatNormalizer> normalizers;

    /** 2. 路由：事件 → 目标处理器 */
    @Resource
    private EventRouter router;

    /** 3. 过滤责任链：幂等 / 开关 / 业务过滤，按 @Order 排序注入 */
    @Resource
    private List<EventFilter> filters;

    public void accept(String rawJson) {
        // ① 归一化：原始报文 → 标准领域事件
        BinlogEvent event = normalize(rawJson);
        if (event == null) {
            return;
        }

        // ② 路由：按表名找到唯一处理器；找不到即丢弃（无意义重试）
        BinlogProcessor processor = router.route(event.getTable());
        if (processor == null) {
            log.warn("表[{}]无处理器，跳过", event.getTable());
            return;
        }

        // ③ 过滤责任链：任一环节拒绝则终止，且各 Filter 互不知晓彼此
        for (EventFilter filter : filters) {
            if (!filter.accept(event)) {
                log.info("被[{}]拦截, table={}, pk={}",
                        filter.getClass().getSimpleName(), event.getTable(), event.getPrimaryKey());
                return;
            }
        }

        // ④ 分发到处理器：只按操作类型回调，处理器不感知前三步
        processor.handle(event);
    }

    private BinlogEvent normalize(String rawJson) {
        return normalizers.stream()
                .filter(n -> n.support(rawJson))
                .findFirst()
                .map(n -> n.normalize(rawJson))
                .orElse(null);
    }
}

@Target(ElementType.TYPE)
@Retention(RetentionPolicy.RUNTIME)
public @interface BinlogTable {
    String[] value();   // 该处理器负责的表名，可多张
}

@Component
public class EventRouter {

    private final Map<String, BinlogProcessor> table2processor = new HashMap<>(16);

    /** 启动时扫描所有 BinlogProcessor，按注解声明的表名建立路由表 */
    public EventRouter(List<BinlogProcessor> processors) {
        for (BinlogProcessor p : processors) {
            BinlogTable anno = p.getClass().getAnnotation(BinlogTable.class);
            if (anno == null) {
                throw new IllegalStateException(p.getClass() + " 缺少 @BinlogTable");
            }
            for (String table : anno.value()) {
                BinlogProcessor exist = table2processor.putIfAbsent(table, p);
                if (exist != null) {
                    throw new IllegalStateException("表[" + table + "]被多个处理器声明");
                }
            }
        }
    }

    public BinlogProcessor route(String table) {
        return table2processor.get(table);
    }
}

public interface EventFilter {
    /** 返回 false 表示拦截，事件不再向后流转 */
    boolean accept(BinlogEvent event);
}

/** ① DDL / 空数据过滤：结构变更不承载业务数据 */
@Order(10)
@Component
public class DdlFilter implements EventFilter {
    @Override
    public boolean accept(BinlogEvent event) {
        return event.getOpType() != null && event.getAfter() != null;
    }
}

/** ② 开关过滤：灰度 / 应急降级时可整体关闭某表消费 */
@Order(20)
@Component
public class SwitchFilter implements EventFilter {

    @Resource
    private BinlogSwitchConfig switchConfig;

    @Override
    public boolean accept(BinlogEvent event) {
        return switchConfig.isEnabled(event.getTable());
    }
}

/** ③ 幂等 + 防乱序：仅当本次 ts 不早于已处理 ts 才放行 */
@Order(30)
@Component
public class IdempotentFilter implements EventFilter {

    @Resource
    private ProcessedTsStore tsStore;   // 记录「主键 → 最近处理时间戳」

    @Override
    public boolean accept(BinlogEvent event) {
        Long last = tsStore.get(event.getTable(), event.getPrimaryKey());
        if (last != null && event.getTs() < last) {
            return false;               // 旧消息，丢弃
        }
        tsStore.put(event.getTable(), event.getPrimaryKey(), event.getTs());
        return true;
    }
}

public interface BinlogProcessor {
    void handle(BinlogEvent event);
}

/** 可选基类：把「每操作的前置判断 + 业务动作」成对固化，非强制继承 */
public abstract class OpTypeProcessor implements BinlogProcessor {

    @Override
    public final void handle(BinlogEvent event) {
        switch (event.getOpType()) {
            case INSERT: if (acceptInsert(event)) onInsert(event); break;
            case UPDATE: if (acceptUpdate(event)) onUpdate(event); break;
            case DELETE: if (acceptDelete(event)) onDelete(event); break;
            default: /* ignore */
        }
    }

    // —— 每个操作独立的「要不要处理」判断，默认全放行，子类按表覆写 ——
    protected boolean acceptInsert(BinlogEvent event) { return true; }
    protected boolean acceptUpdate(BinlogEvent event) { return true; }
    protected boolean acceptDelete(BinlogEvent event) { return true; }

    // —— 每个操作的业务动作，默认空实现，子类按需覆写 ——
    protected void onInsert(BinlogEvent event) {}
    protected void onUpdate(BinlogEvent event) {}
    protected void onDelete(BinlogEvent event) {}
}

@BinlogTable("order_status")
@Component
public class OrderStatusProcessor extends OpTypeProcessor {

    @Resource
    private OrderSyncService orderSyncService;

    /** 新增：只同步目标业务类型的订单 */
    @Override
    protected boolean acceptInsert(BinlogEvent event) {
        return isTargetBiz(event.getAfter());
    }

    @Override
    protected void onInsert(BinlogEvent event) {
        orderSyncService.sync(toOrder(event.getAfter()));
    }

    /** 更新：目标类型 且 状态确实发生变化才处理（避免无意义写下游） */
    @Override
    protected boolean acceptUpdate(BinlogEvent event) {
        return isTargetBiz(event.getAfter()) && statusChanged(event);
    }

    @Override
    protected void onUpdate(BinlogEvent event) {
        orderSyncService.sync(toOrder(event.getAfter()));
    }

    // 未覆写 acceptDelete → 默认放行；也可覆写为 return false 表示本表不处理删除
    @Override
    protected void onDelete(BinlogEvent event) {
        orderSyncService.remove(event.getPrimaryKey());
    }

    private boolean isTargetBiz(JSONObject after) {
        return "NORMAL".equals(after.getString("biz_type"));
    }

    private boolean statusChanged(BinlogEvent event) {
        return event.getBefore() != null
                && !Objects.equals(event.getBefore().getString("status"),
                                   event.getAfter().getString("status"));
    }

    private Order toOrder(JSONObject after) {
        return after.toJavaObject(Order.class);
    }
}