灰度发布：把新功能的开关按部门切开

import sqlite3

SCHEMA = """
CREATE TABLE IF NOT EXISTS feature_flag (
    flag_key   TEXT    NOT NULL,
    dept_code  TEXT    NOT NULL,
    enabled    INTEGER NOT NULL DEFAULT 0,
    revision   INTEGER NOT NULL DEFAULT 1,
    updated_by TEXT    NOT NULL,
    updated_at INTEGER NOT NULL,
    PRIMARY KEY (flag_key, dept_code)
);
"""


class FlagStore:
    def __init__(self, db_path):
        self.conn = sqlite3.connect(db_path)
        self.conn.executescript(SCHEMA)

    def is_enabled(self, flag_key, dept_code):
        node = dept_code
        while node:
            row = self.conn.execute(
                "SELECT enabled, revision FROM feature_flag "
                "WHERE flag_key = ? AND dept_code = ?",
                (flag_key, node),
            ).fetchone()
            if row is not None:
                return bool(row[0]), row[1]
            node = node.rpartition(".")[0]
        return False, 0

    def set_enabled(self, flag_key, dept_code, enabled, operator):
        self.conn.execute(
            "INSERT INTO feature_flag "
            "(flag_key, dept_code, enabled, revision, updated_by, updated_at) "
            "VALUES (?, ?, ?, 1, ?, strftime('%s','now')) "
            "ON CONFLICT(flag_key, dept_code) DO UPDATE SET "
            "enabled = excluded.enabled, "
            "revision = feature_flag.revision + 1, "
            "updated_by = excluded.updated_by, "
            "updated_at = excluded.updated_at",
            (flag_key, dept_code, int(enabled), operator),
        )
        self.conn.commit()

# 入口只做三件事：定部门、判开关、通过才往下走
def handle_assist(req, flags, directory, pipeline):
    dept = directory.dept_of(req.user_id)
    if not dept:
        return legacy_entry(req, None, reason="dept_unknown")

    enabled, revision = flags.is_enabled("ai.assist", dept.code)
    if not enabled:
        return legacy_entry(req, dept.code, reason="flag_off")

    docs = pipeline.retrieve(req.question, dept)
    answer = pipeline.generate(req.question, docs)
    pipeline.audit(req.user_id, dept.code, revision, docs)
    return answer

# 关闭不是拒绝，是回到旧入口，响应形态与未上线时保持一致
def legacy_entry(req, dept_code, reason):
    log.info(
        "assist fallback user=%s dept=%s reason=%s",
        req.user_id, dept_code, reason,
    )
    resp = make_response(redirect("/assist/classic"))
    resp.headers["X-Assist-Mode"] = "legacy"
    return resp


def health_of_flags(flags, depts):
    # 收回后用来确认没有残留流量还在走新链路
    return {
        d.code: flags.is_enabled("ai.assist", d.code)[0]
        for d in depts
    }

import time


# 收回要秒级生效，TTL 只给几秒，配合改动时的主动失效
class FlagCache:
    def __init__(self, store, ttl=5.0):
        self._store = store
        self._ttl = ttl
        self._memo = {}

    def is_enabled(self, key, dept):
        now = time.monotonic()
        hit = self._memo.get((key, dept))
        if hit is not None and hit[0] > now:
            return hit[1], hit[2]
        enabled, revision = self._store.is_enabled(key, dept)
        self._memo[(key, dept)] = (now + self._ttl, enabled, revision)
        return enabled, revision

    def invalidate(self, key=None):
        if key is None:
            self._memo.clear()
            return
        for k in [k for k in self._memo if k[0] == key]:
            self._memo.pop(k, None)