Files
caowangrenbo 0fffdd5ccb test: 集成测试改用真 LLDAP 容器;修复两处物化缺陷
- 集成测试重构:syncer e2e/materialize、ldaplib 读路径、lldapgql 全改用本地
  LLDAP 容器(LDAP 3890 / GraphQL /api/graphql 17170),删旧 OpenLDAP 断言
- 修复 1:ActiveUsers 不填 DepartmentIDs(存于关系表),BuildFromMirror 拿到空成员;
  新增 ActiveUsersWithDepts,物化 plan 用之
- 修复 2:deleted 用户重新可见后不回 active,无法重建;RevertToActive 纳入 deleted,
  快照 reconcile 对其置回 active + 作废 token
- 关键安全回归单测保留并配变异验证:未确认不删、非受管不删、admin 永不移出、
  删除最后阶段、绝不碰密码。全量单测+vet+集成全绿
2026-09-29 23:53:24 +08:00

354 lines
12 KiB
Go
Raw Permalink Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
package syncer
// 快照 runner:拉取飞书全量 → 事务写入 SQLite 镜像 → 离职生命周期 reconcile。
//
// 这是两阶段架构的「第一段」:它只把「飞书当前有什么」落到镜像,并把离职流程
// 推进到「已发卡片待人工确认」为止。物化(写 LLDAP)由另一条链路承担,本 runner
// 不碰 LLDAP,也不做任何写入决策。
import (
"context"
"fmt"
"log/slog"
"sort"
"strings"
"time"
"idsync/internal/config"
"idsync/internal/feishu"
"idsync/internal/mapping"
"idsync/internal/store"
)
// SnapshotOptions 是快照 runner 的运行参数。
type SnapshotOptions struct {
// WebhookURL 是离职确认卡片的自定义机器人地址。
WebhookURL string
// ConfirmBaseURL 是确认端点的对外 base-url(可被公网/反代访问)。
ConfirmBaseURL string
// ReNotifyInterval 是离职提醒的最短间隔,缺省 24h,防刷屏。
ReNotifyInterval time.Duration
// TokenTTL 是一次性确认 token 的有效期,缺省 72h。
TokenTTL time.Duration
}
func (o SnapshotOptions) withDefaults() SnapshotOptions {
if o.ReNotifyInterval <= 0 {
o.ReNotifyInterval = 24 * time.Hour
}
if o.TokenTTL <= 0 {
o.TokenTTL = 72 * time.Hour
}
return o
}
// SnapshotRunResult 是一次快照执行的统计结果。
type SnapshotRunResult struct {
Users int
Departments int
Skipped int
LeaveCandidates int
PendingConfirm int
Reinstated int
CardsSent int
Warnings []string
}
// SnapshotRunner 编排快照流程。
type SnapshotRunner struct {
cfg *config.Config
feishu *feishu.Client
store *store.Store
log *slog.Logger
opts SnapshotOptions
}
// NewSnapshotRunner 构造快照 runner。
func NewSnapshotRunner(cfg *config.Config, f *feishu.Client, st *store.Store, log *slog.Logger, opts SnapshotOptions) *SnapshotRunner {
return &SnapshotRunner{cfg: cfg, feishu: f, store: st, log: log, opts: opts.withDefaults()}
}
// RunSnapshot 执行一轮快照:拉取、落库、推进离职流程。
func (r *SnapshotRunner) RunSnapshot(ctx context.Context) (*SnapshotRunResult, error) {
snap, err := r.feishu.Fetch(ctx, r.cfg.Feishu.DeptList)
if err != nil {
return nil, fmt.Errorf("读取飞书快照失败: %w", err)
}
r.log.Info("已读取飞书快照",
"departments", len(snap.Departments), "users", len(snap.Users),
"skipped_no_email", len(snap.SkippedNoEmail), "duplicates", len(snap.SkippedDuplicate))
mirror := r.buildMirror(snap)
if err := r.store.ReplaceMirror(mirror); err != nil {
return nil, fmt.Errorf("写入 SQLite 镜像失败: %w", err)
}
present := make(map[string]bool, len(mirror.Users))
for _, u := range mirror.Users {
present[u.UID] = true
}
res := &SnapshotRunResult{
Users: len(mirror.Users),
Departments: len(snap.Departments),
Skipped: len(mirror.Skipped),
}
if err := r.reconcile(ctx, present, res); err != nil {
return res, err
}
return res, nil
}
// buildMirror 把飞书快照转成镜像存储层需要的输入。
//
// 部门组 DN 用稳定索引派生(mapping.DeptGroupDN),可读中文名只落 name 列供展示。
func (r *SnapshotRunner) buildMirror(snap *feishu.Snapshot) *store.MirrorSnapshot {
deptName := make(map[string]string, len(snap.Departments))
for _, d := range snap.Departments {
deptName[d.ID] = d.Name
}
depts := make([]*store.MirrorDepartment, 0, len(snap.Departments))
for _, d := range snap.Departments {
depts = append(depts, &store.MirrorDepartment{
DeptID: d.ID, Name: d.Name, DN: mapping.DeptGroupDN(d.ID, r.cfg.Ldap.BaseDN),
ParentID: d.ParentID, Level: deptLevel(d.ID, snap.Departments), State: store.StateActive,
})
}
users := make([]*store.MirrorUser, 0, len(snap.Users))
for _, u := range snap.Users {
uid, ok := feishu.LoginName(u)
if !ok {
continue // 已在 SkippedNoEmail 报告中,不必重复
}
users = append(users, toMirrorUser(u, uid, deptName))
}
// 缺少企业邮箱的用户以 union_id 作为 uid 落一条 skipped 记录,供 WebUI 列出。
skipped := make([]*store.MirrorUser, 0, len(snap.SkippedNoEmail))
for _, u := range snap.SkippedNoEmail {
skipped = append(skipped, &store.MirrorUser{
UID: u.UnionID,
UnionID: u.UnionID,
UserID: u.UserID,
DisplayName: feishu.DisplayName(u),
DepartmentIDs: u.DepartmentIDs,
State: store.StateActive,
})
}
return &store.MirrorSnapshot{Users: users, Departments: depts, Skipped: skipped}
}
// deptLevel 计算部门在树中的层级,根为 0。带步数上限防环。
func deptLevel(id string, depts []*feishu.Department) int {
byID := make(map[string]*feishu.Department, len(depts))
for _, d := range depts {
byID[d.ID] = d
}
depth := 0
cur := id
for i := 0; i < 64; i++ {
d, ok := byID[cur]
if !ok || d.ParentID == "" || d.ParentID == "0" {
return depth
}
if _, ok := byID[d.ParentID]; !ok {
return depth // 父不在同步范围内,视为根
}
depth++
cur = d.ParentID
}
return depth
}
// toMirrorUser 把飞书用户域模型转成镜像用户。字段语义与 mapping 一致,
// business_category 展开为部门中文名便于人工阅读。
func toMirrorUser(u *feishu.User, uid string, deptName map[string]string) *store.MirrorUser {
var bc []string
for _, id := range u.DepartmentIDs {
if n, ok := deptName[id]; ok && n != "" {
bc = append(bc, n)
}
}
sort.Strings(bc)
return &store.MirrorUser{
UID: uid,
UnionID: u.UnionID,
UserID: u.UserID,
Email: u.EnterpriseEmail,
DisplayName: firstNonEmpty(u.Name, u.EnName, uid),
GivenName: firstNonEmpty(u.EnName, u.Name, uid),
SN: firstName(u),
Mobile: u.Mobile,
EmployeeNumber: u.EmployeeNo,
DepartmentNumber: u.JobTitle,
PostalAddress: u.WorkStation,
BusinessCategory: strings.Join(bc, ","),
EmployeeType: mapping.EmployeeTypeActive,
Description: firstNonEmpty(u.Name, u.EnName, uid),
DepartmentIDs: u.DepartmentIDs,
State: store.StateActive,
}
}
// reconcile 处理镜像里每个用户的生命周期:本轮在飞书可见者撤销离职流程;
// 缺席者在职用户做机器确认,确认后进发卡流程;已在流程内者按节流策略补发提醒。
//
// 这里的每个状态转移都基于 store 的受校验状态机,任何非法转移会被结构性拦下。
func (r *SnapshotRunner) reconcile(ctx context.Context, present map[string]bool, res *SnapshotRunResult) error {
all, err := r.store.AllMirrorUsers()
if err != nil {
return err
}
for _, u := range all {
if u.Skipped {
continue
}
if present[u.UID] {
// 本轮在飞书可见:撤销离职流程并作废未用的 token。
switch u.State {
case store.StateLeaveCandidate, store.StatePendingConfirm,
store.StateConfirmedLeave, store.StateDeleted:
// 含已删除用户重新入职:置回 active,物化以「重建」处理。
if err := r.store.RevertToActive(u.UID); err != nil {
return fmt.Errorf("撤销离职流程 uid=%s: %w", u.UID, err)
}
if err := r.store.RevokeTokensForUser(u.UID); err != nil {
return fmt.Errorf("作废离职 token uid=%s: %w", u.UID, err)
}
res.Reinstated++
r.log.Info("用户重新可见,已撤销离职流程", "uid", u.UID)
}
continue
}
// 本轮缺席。
switch u.State {
case store.StateActive:
if err := r.handleAbsentCandidate(ctx, u, res); err != nil {
return err
}
case store.StateLeaveCandidate:
// 上轮已机器确认离职、卡片尚未发出,本轮补发。
if err := r.sendResignCard(ctx, u); err != nil {
res.Warnings = append(res.Warnings, fmt.Sprintf("uid=%s 离职卡片发送失败,稍后重试: %v", u.UID, err))
continue
}
res.CardsSent++
if err := r.store.ToPendingConfirm(u.UID); err != nil {
return fmt.Errorf("转入待确认 uid=%s: %w", u.UID, err)
}
res.PendingConfirm++
case store.StatePendingConfirm:
if err := r.renotify(ctx, u, res); err != nil {
return err
}
case store.StateConfirmedLeave, store.StateDeleted:
// 已人工确认、等待物化删除一组:不再打扰。
}
}
return nil
}
// handleAbsentCandidate 处理「在职但本轮缺席」的用户:先做飞书机器确认,
// 确认离职才进入发卡流程;未确认或缺 union_id 时保守不动作并告警。
func (r *SnapshotRunner) handleAbsentCandidate(ctx context.Context, u *store.MirrorUser, res *SnapshotRunResult) error {
if u.UnionID == "" {
res.Warnings = append(res.Warnings, fmt.Sprintf("用户 %s 缺席但无 union_id,无法机器确认离职,暂不动作", u.UID))
return nil
}
resigned, err := r.feishu.CheckResigned(ctx, []string{u.UnionID})
if err != nil {
res.Warnings = append(res.Warnings, fmt.Sprintf("确认 uid=%s 离职状态失败,本轮不动: %v", u.UID, err))
return nil
}
if len(resigned) == 0 {
res.Warnings = append(res.Warnings, fmt.Sprintf("用户 %s 缺席但飞书未确认离职(可能因通讯录权限范围收窄),暂不动作", u.UID))
return nil
}
if err := r.store.ToLeaveCandidate(u.UID); err != nil {
return fmt.Errorf("标记离职候选 uid=%s: %w", u.UID, err)
}
res.LeaveCandidates++
if err := r.sendResignCard(ctx, u); err != nil {
res.Warnings = append(res.Warnings, fmt.Sprintf("uid=%s 离职卡片发送失败,稍后重试: %v", u.UID, err))
return nil
}
res.CardsSent++
if err := r.store.ToPendingConfirm(u.UID); err != nil {
return fmt.Errorf("转入待确认 uid=%s: %w", u.UID, err)
}
res.PendingConfirm++
return nil
}
// renotify 对「已待确认但长期未点」的用户按节流间隔重发提醒,避免每轮刷屏。
// 超时后发一张新卡片并作废旧 token,卡内的新 token 重新起效。
func (r *SnapshotRunner) renotify(ctx context.Context, u *store.MirrorUser, res *SnapshotRunResult) error {
last, err := r.store.LastPendingToken(u.UID)
if err != nil {
return err
}
if last != nil && time.Since(last.CreatedAt) < r.opts.ReNotifyInterval {
return nil // 节流:距上次提醒不足间隔,不重发、不打扰
}
if err := r.sendResignCard(ctx, u); err != nil {
res.Warnings = append(res.Warnings, fmt.Sprintf("uid=%s 离职提醒发送失败: %v", u.UID, err))
return nil
}
res.CardsSent++
return nil
}
// sendResignCard 为某用户发一张离职确认卡片:生成一次性 token(作废旧 token,
// 保证任一时刻至多一张有效的确认链接,减少不确定的凭证)、构造确认 URL、发送。
func (r *SnapshotRunner) sendResignCard(ctx context.Context, u *store.MirrorUser) error {
// 先作废旧 token 再发新 token,避免卡片按钮连着多张「都能确认」的链接。
if err := r.store.RevokeTokensForUser(u.UID); err != nil {
return err
}
tok, err := r.store.CreateToken(u.UID, r.opts.TokenTTL)
if err != nil {
return err
}
confirmURL := config.ConfirmURL(r.opts.ConfirmBaseURL, u.UID, tok.Token)
payload, err := feishu.BuildResignCardPayload(&feishu.ResignCard{
DisplayName: u.DisplayName,
UID: u.UID,
Reason: "飞书通讯录中已不可见,经机器确认已离职,请人工复核后确认。",
ConfirmURL: confirmURL,
})
if err != nil {
return err
}
return feishu.SendResignCard(ctx, r.opts.WebhookURL, payload)
}
// firstName 为 SN 提供非空值(与映射语义一致)。
func firstName(u *feishu.User) string {
if n := strings.TrimSpace(u.Name); n != "" {
return n
}
if n := strings.TrimSpace(u.EnName); n != "" {
return n
}
return u.UnionID
}
func firstNonEmpty(vals ...string) string {
for _, v := range vals {
if strings.TrimSpace(v) != "" {
return v
}
}
return ""
}