- 集成测试重构: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+集成全绿
354 lines
12 KiB
Go
354 lines
12 KiB
Go
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 ""
|
||
} |