diff --git a/config.go b/config.go new file mode 100644 index 0000000..0a9518b --- /dev/null +++ b/config.go @@ -0,0 +1,33 @@ +package main + +import ( + "encoding/json" + "os" + "sync" + "time" +) + +// Config 对应 config.json 结构 +type Config struct { + SecretToken string `json:"secret_token"` + Projects map[string]map[string]string `json:"projects"` + DeviceKeys []string `json:"device_keys"` +} + +var ( + globalConfig Config + // pendingDebounceMap 防抖窗口内的待执行任务,key: "repoName:ref", value: *debounceEntry + pendingDebounceMap sync.Map + // 防抖冷却时间限制:5 分钟 + debounceDuration = 5 * time.Minute + // bark 推送消息 + pushURL = "https://bark.maimaicuizhiji.top/push" +) + +func loadConfig(path string) error { + file, err := os.ReadFile(path) + if err != nil { + return err + } + return json.Unmarshal(file, &globalConfig) +} diff --git a/main.go b/main.go index 38578a3..0088bbc 100644 --- a/main.go +++ b/main.go @@ -1,240 +1,14 @@ package main import ( - "bytes" - "encoding/json" "flag" "fmt" - "io" "log" - "net/http" - "os" - "os/exec" - "sync" "time" "github.com/gin-gonic/gin" ) -// Config 对应 config.json 结构 -type Config struct { - SecretToken string `json:"secret_token"` - Projects map[string]map[string]string `json:"projects"` - DeviceKeys []string `json:"device_keys"` -} - -// GitLabPayload 简化版的 GitLab Push Webhook 数据结构 -type GitLabPayload struct { - Ref string `json:"ref"` - After string `json:"after"` - Project struct { - PathWithNamespace string `json:"path_with_namespace"` - } `json:"project"` -} - -// BarkPushPayload 定义 Bark 推送的请求 JSON 结构(单 device_key) -type BarkPushPayload struct { - Title string `json:"title"` - Body string `json:"body"` - Group string `json:"group,omitempty"` - DeviceKey string `json:"device_key"` - IsArchive string `json:"isArchive,omitempty"` - TTL int `json:"ttl,omitempty"` -} - -// SendNotification 依次对每个 device_key 发送 Bark 通知,遇错即返回 -func SendNotification(pushURL string, deviceKeys []string, title, body string) error { - for _, key := range deviceKeys { - if err := sendOne(pushURL, key, title, body); err != nil { - return err - } - } - return nil -} - -// sendOne 向单个 device_key 发送推送 -func sendOne(pushURL, deviceKey, title, body string) error { - payload := BarkPushPayload{ - Title: title, - Body: body, - Group: "Model Development Platform", - DeviceKey: deviceKey, - IsArchive: "1", - TTL: 3600, - } - - jsonBytes, err := json.Marshal(payload) - if err != nil { - return fmt.Errorf("JSON 序列化失败: %w", err) - } - - req, err := http.NewRequest("POST", pushURL, bytes.NewBuffer(jsonBytes)) - if err != nil { - return fmt.Errorf("创建请求失败: %w", err) - } - req.Header.Set("Content-Type", "application/json; charset=utf-8") - - client := &http.Client{Timeout: 5 * time.Second} - resp, err := client.Do(req) - if err != nil { - return fmt.Errorf("发送推送请求失败: %w", err) - } - defer resp.Body.Close() - - if resp.StatusCode >= 400 { - bodyBytes, _ := io.ReadAll(io.LimitReader(resp.Body, 512)) - bodyText := string(bytes.TrimSpace(bodyBytes)) - if bodyText != "" { - return fmt.Errorf("推送服务器返回异常状态码: %d, 响应: %s", resp.StatusCode, bodyText) - } - return fmt.Errorf("推送服务器返回异常状态码: %d", resp.StatusCode) - } - - return nil -} - -var ( - globalConfig Config - // lastExecutionMap 用于记录某个仓库+分支的上一次触发时间 key: "repoName:ref", value: time.Time - lastExecutionMap sync.Map - // 防抖冷却时间限制:5 分钟 - debounceDuration = 5 * time.Minute - // bark 推送消息 - pushURL = "https://bark.maimaicuizhiji.top/push" -) - -func loadConfig(path string) error { - file, err := os.ReadFile(path) - if err != nil { - return err - } - return json.Unmarshal(file, &globalConfig) -} - -// 异步执行 Shell 脚本 -func runScript(scriptPath, repoName, ref, commitID string) { - log.Printf("[INFO] 开始执行脚本: %s (仓库: %s, 分支: %s)\n", scriptPath, repoName, ref) - - cmd := exec.Command("bash", scriptPath) - - // 传入环境变量,脚本内可读取 - cmd.Env = append(os.Environ(), - fmt.Sprintf("GIT_REPO_NAME=%s", repoName), - fmt.Sprintf("GIT_REF=%s", ref), - fmt.Sprintf("GIT_COMMIT_ID=%s", commitID), - ) - - // 捕获输出 - output, err := cmd.CombinedOutput() - if err != nil { - log.Printf("[ERROR] 脚本执行失败 [%s]: %v\n输出:\n%s\n", scriptPath, err, string(output)) - title := fmt.Sprintf("%s更新失败", repoName) - body := fmt.Sprintf("分支: %s\nCommit: %s\n服务器已接收到 Push,更新失败", ref, commitID[:7]) - if err := SendNotification(pushURL, globalConfig.DeviceKeys, title, body); err != nil { - log.Printf("[WARNING] 推送通知失败: %v\n", err) - } else { - log.Println("[INFO] 推送通知发送成功") - } - return - } - - log.Printf("[SUCCESS] 脚本执行成功 [%s]\n输出:\n%s\n", scriptPath, string(output)) - - title := fmt.Sprintf("%s更新成功", repoName) - body := fmt.Sprintf("分支: %s\nCommit: %s\n服务器已接收到 Push,更新成功", ref, commitID[:7]) - if err := SendNotification(pushURL, globalConfig.DeviceKeys, title, body); err != nil { - log.Printf("[WARNING] 推送通知失败: %v\n", err) - } else { - log.Println("[INFO] 推送通知发送成功") - } -} - -func handleWebhook(c *gin.Context) { - // 1. 校验 GitLab Secret Token Header - clientToken := c.GetHeader("X-Gitlab-Token") - if globalConfig.SecretToken != "" && clientToken != globalConfig.SecretToken { - c.JSON(http.StatusUnauthorized, gin.H{"error": "Invalid Secret Token"}) - return - } - - // 2. 校验 Event 类型 - eventType := c.GetHeader("X-Gitlab-Event") - if eventType != "Push Hook" { - c.JSON(http.StatusOK, gin.H{"status": "ignored", "reason": "Only Push Hook is handled"}) - return - } - - // 3. 解析 Body JSON - var payload GitLabPayload - if err := c.ShouldBindJSON(&payload); err != nil { - c.JSON(http.StatusBadRequest, gin.H{"error": "Invalid JSON Payload"}) - return - } - - repoName := payload.Project.PathWithNamespace - ref := payload.Ref - commitID := payload.After - - if repoName == "" || ref == "" { - c.JSON(http.StatusBadRequest, gin.H{"error": "Missing repo or ref in payload"}) - return - } - - // 4. 匹配脚本路径 - repoConfig, exists := globalConfig.Projects[repoName] - if !exists { - c.JSON(http.StatusOK, gin.H{"status": "ignored", "message": "No script config for repository " + repoName}) - return - } - - scriptPath, exists := repoConfig[ref] - if !exists { - c.JSON(http.StatusOK, gin.H{"status": "ignored", "message": fmt.Sprintf("No script config for %s on %s", repoName, ref)}) - return - } - - // 5. 校验脚本文件是否存在 - if _, err := os.Stat(scriptPath); os.IsNotExist(err) { - c.JSON(http.StatusInternalServerError, gin.H{"error": "Script file not found: " + scriptPath}) - return - } - - // 6. 核心逻辑:5 分钟防抖检查 - lockKey := fmt.Sprintf("%s:%s", repoName, ref) - now := time.Now() - - if lastTimeVal, loaded := lastExecutionMap.Load(lockKey); loaded { - lastTime := lastTimeVal.(time.Time) - elapsed := now.Sub(lastTime) - if elapsed < debounceDuration { - remainingSeconds := int((debounceDuration - elapsed).Seconds()) - msg := fmt.Sprintf("触发太频繁,防抖拦截中(5分钟限制),剩余等待时间: %d 秒", remainingSeconds) - log.Printf("[DEBOUNCE] 跳过执行 [%s] -> %s\n", lockKey, msg) - - c.JSON(http.StatusOK, gin.H{ - "status": "debounced", - "message": msg, - "remaining_seconds": remainingSeconds, - }) - return - } - } - - // 更新上一次触发时间 - lastExecutionMap.Store(lockKey, now) - - // 7. 开启 goroutine 异步执行脚本 - go runScript(scriptPath, repoName, ref, commitID) - - // 8. 立即回应 200 给 GitLab - c.JSON(http.StatusOK, gin.H{ - "status": "triggered", - "repository": repoName, - "ref": ref, - "script": scriptPath, - }) -} - func main() { portFlag := flag.Int("port", 8000, "服务监听端口") flag.Parse() diff --git a/notify.go b/notify.go new file mode 100644 index 0000000..e812d59 --- /dev/null +++ b/notify.go @@ -0,0 +1,71 @@ +package main + +import ( + "bytes" + "encoding/json" + "fmt" + "io" + "net/http" + "time" +) + +// BarkPushPayload 定义 Bark 推送的请求 JSON 结构(单 device_key) +type BarkPushPayload struct { + Title string `json:"title"` + Body string `json:"body"` + Group string `json:"group,omitempty"` + DeviceKey string `json:"device_key"` + IsArchive string `json:"isArchive,omitempty"` + TTL int `json:"ttl,omitempty"` +} + +// SendNotification 依次对每个 device_key 发送 Bark 通知,遇错即返回 +func SendNotification(pushURL string, deviceKeys []string, title, body string) error { + for _, key := range deviceKeys { + if err := sendOne(pushURL, key, title, body); err != nil { + return err + } + } + return nil +} + +// sendOne 向单个 device_key 发送推送 +func sendOne(pushURL, deviceKey, title, body string) error { + payload := BarkPushPayload{ + Title: title, + Body: body, + Group: "Model Development Platform", + DeviceKey: deviceKey, + IsArchive: "1", + TTL: 3600, + } + + jsonBytes, err := json.Marshal(payload) + if err != nil { + return fmt.Errorf("JSON 序列化失败: %w", err) + } + + req, err := http.NewRequest("POST", pushURL, bytes.NewBuffer(jsonBytes)) + if err != nil { + return fmt.Errorf("创建请求失败: %w", err) + } + req.Header.Set("Content-Type", "application/json; charset=utf-8") + + client := &http.Client{Timeout: 5 * time.Second} + resp, err := client.Do(req) + if err != nil { + return fmt.Errorf("发送推送请求失败: %w", err) + } + defer resp.Body.Close() + + if resp.StatusCode >= 400 { + bodyBytes, _ := io.ReadAll(io.LimitReader(resp.Body, 512)) + bodyText := string(bytes.TrimSpace(bodyBytes)) + if bodyText != "" { + return fmt.Errorf("推送服务器返回异常状态码: %d, 响应: %s", resp.StatusCode, bodyText) + } + return fmt.Errorf("推送服务器返回异常状态码: %d", resp.StatusCode) + } + + return nil +} diff --git a/runner.go b/runner.go new file mode 100644 index 0000000..439260b --- /dev/null +++ b/runner.go @@ -0,0 +1,121 @@ +package main + +import ( + "bufio" + "context" + "errors" + "fmt" + "io" + "log" + "os" + "os/exec" + "strings" + "sync" + "time" +) + +// scriptTimeout 脚本执行超时上限(15 分钟) +const scriptTimeout = 15 * time.Minute + +// shortCommit 安全地取 commit 短 ID,避免空值/短值切片越界 panic +func shortCommit(commitID string) string { + if len(commitID) > 7 { + return commitID[:7] + } + if commitID == "" { + return "unknown" + } + return commitID +} + +// streamLog 逐行读取并实时打印脚本输出,直到 r 关闭 +func streamLog(r io.Reader, tag string) { + br := bufio.NewReader(r) + for { + line, err := br.ReadString('\n') + if line != "" { + log.Printf("[SCRIPT %s] %s", tag, strings.TrimRight(line, "\r\n")) + } + if err != nil { + return + } + } +} + +// notifyResult 发送执行结果通知 +func notifyResult(ok bool, repoName, ref, commitID string, cost time.Duration, reason string) { + state := "成功" + if !ok { + state = "失败" + } + title := fmt.Sprintf("%s更新%s", repoName, state) + body := fmt.Sprintf("分支: %s\nCommit: %s\n耗时: %s\n原因: %s\n服务器已接收到 Push,更新%s", + ref, shortCommit(commitID), cost.Round(time.Second), reason, state) + + if err := SendNotification(pushURL, globalConfig.DeviceKeys, title, body); err != nil { + log.Printf("[WARNING] 推送通知失败: %v", err) + return + } + log.Println("[INFO] 推送通知发送成功") +} + +// runScript 异步执行 Shell 脚本,输出实时流式打印,超时 15 分钟 +func runScript(scriptPath, repoName, ref, commitID string) { + start := time.Now() + tag := fmt.Sprintf("%s:%s", repoName, ref) + log.Printf("[INFO] 开始执行脚本: %s (仓库: %s, 分支: %s) 超时上限 %s", + scriptPath, repoName, ref, scriptTimeout) + + ctx, cancel := context.WithTimeout(context.Background(), scriptTimeout) + defer cancel() + + cmd := exec.CommandContext(ctx, "bash", scriptPath) + // 超时/取消后最多再给 10 秒让进程优雅退出;超时则强制 SIGKILL + cmd.WaitDelay = 10 * time.Second + + // 传入环境变量,脚本内可读取 + cmd.Env = append(os.Environ(), + fmt.Sprintf("GIT_REPO_NAME=%s", repoName), + fmt.Sprintf("GIT_REF=%s", ref), + fmt.Sprintf("GIT_COMMIT_ID=%s", commitID), + ) + + // stdout / stderr 合并到同一个管道,保持原有交错顺序 + pr, pw := io.Pipe() + cmd.Stdout = pw + cmd.Stderr = pw + + if err := cmd.Start(); err != nil { + pw.Close() + pr.Close() + log.Printf("[ERROR] 脚本启动失败 [%s]: %v", scriptPath, err) + notifyResult(false, repoName, ref, commitID, time.Since(start), fmt.Sprintf("启动失败: %v", err)) + return + } + + var wg sync.WaitGroup + wg.Go(func() { + streamLog(pr, tag) + }) + + err := cmd.Wait() // 进程退出 + 内部 copy goroutine 结束 + pw.Close() // 关闭写端,读端收到 EOF + wg.Wait() // 等日志全部刷完再往下走 + + cost := time.Since(start) + + // 先判断超时(ctx.Err() 在超时/取消时非 nil) + switch { + case errors.Is(ctx.Err(), context.DeadlineExceeded): + log.Printf("[TIMEOUT] 脚本执行超时 [%s] 已运行 %s 上限 %s", scriptPath, cost.Round(time.Second), scriptTimeout) + notifyResult(false, repoName, ref, commitID, cost, "执行超时,已强制终止") + return + case err != nil: + log.Printf("[ERROR] 脚本执行失败 [%s] 耗时 %s: %v", scriptPath, cost.Round(time.Millisecond), err) + notifyResult(false, repoName, ref, commitID, cost, err.Error()) + return + } + + log.Printf("[SUCCESS] 脚本执行成功 [%s] 耗时 %s", scriptPath, cost.Round(time.Millisecond)) + notifyResult(true, repoName, ref, commitID, cost, "执行成功") +} diff --git a/webhook.go b/webhook.go new file mode 100644 index 0000000..61b119d --- /dev/null +++ b/webhook.go @@ -0,0 +1,175 @@ +package main + +import ( + "fmt" + "log" + "net/http" + "os" + "sync" + "time" + + "github.com/gin-gonic/gin" +) + +// GitLabPayload 简化版的 GitLab Push Webhook 数据结构 +type GitLabPayload struct { + Ref string `json:"ref"` + After string `json:"after"` + Project struct { + PathWithNamespace string `json:"path_with_namespace"` + } `json:"project"` +} + +// debounceEntry 单个 key 的防抖状态。 +// lastExec:上次实际执行时刻(首次或 timer fire)。 +// timer:当前活动定时器,为 nil 表示已 fire 或未安排。 +// pendingCommit:防抖窗口内累计的最新 commitID,timer fire 时用它执行。 +// 其余字段缓存以便 timer fire 回调无需外部参数。 +type debounceEntry struct { + mu sync.Mutex + lastExec time.Time + timer *time.Timer + pendingCommit string + + scriptPath string + repoName string + ref string +} + +// scheduleLocked 续命或新建防抖定时器(必须持有 entry.mu)。 +// fireAt = now + debounceDuration,pendingCommit 由调用方预先设置。 +// fire 闭包内通过指针身份做代际校验:若已被新定时器顶替,本次不执行。 +func (e *debounceEntry) scheduleLocked(lockKey string) { + if e.timer != nil { + e.timer.Stop() + } + + e.timer = time.AfterFunc(debounceDuration, func() { + e.mu.Lock() + if e.timer == nil { + e.mu.Unlock() + return + } + fired := e.timer + e.mu.Unlock() + + // 代际校验:再次拿锁确认 fired 仍是当前定时器 + e.mu.Lock() + if fired == nil || e.timer != fired { + e.mu.Unlock() + return // 已被新 push 顶替 + } + e.timer = nil + e.lastExec = time.Now() + finalCommit := e.pendingCommit + sp, rn, rf := e.scriptPath, e.repoName, e.ref + pendingDebounceMap.Delete(lockKey) + e.mu.Unlock() + + log.Printf("[DEBOUNCE-FIRE] 防抖窗口结束,补刀执行 [%s] (使用最新 commit %s)\n", + lockKey, shortCommit(finalCommit)) + go runScript(sp, rn, rf, finalCommit) + }) +} + +func handleWebhook(c *gin.Context) { + // 1. 校验 GitLab Secret Token Header + clientToken := c.GetHeader("X-Gitlab-Token") + if globalConfig.SecretToken != "" && clientToken != globalConfig.SecretToken { + c.JSON(http.StatusUnauthorized, gin.H{"error": "Invalid Secret Token"}) + return + } + + // 2. 校验 Event 类型 + eventType := c.GetHeader("X-Gitlab-Event") + if eventType != "Push Hook" { + c.JSON(http.StatusOK, gin.H{"status": "ignored", "reason": "Only Push Hook is handled"}) + return + } + + // 3. 解析 Body JSON + var payload GitLabPayload + if err := c.ShouldBindJSON(&payload); err != nil { + c.JSON(http.StatusBadRequest, gin.H{"error": "Invalid JSON Payload"}) + return + } + + repoName := payload.Project.PathWithNamespace + ref := payload.Ref + commitID := payload.After + + if repoName == "" || ref == "" { + c.JSON(http.StatusBadRequest, gin.H{"error": "Missing repo or ref in payload"}) + return + } + + // 4. 匹配脚本路径 + repoConfig, exists := globalConfig.Projects[repoName] + if !exists { + c.JSON(http.StatusOK, gin.H{"status": "ignored", "message": "No script config for repository " + repoName}) + return + } + + scriptPath, exists := repoConfig[ref] + if !exists { + c.JSON(http.StatusOK, gin.H{"status": "ignored", "message": fmt.Sprintf("No script config for %s on %s", repoName, ref)}) + return + } + + // 5. 校验脚本文件是否存在 + if _, err := os.Stat(scriptPath); os.IsNotExist(err) { + c.JSON(http.StatusInternalServerError, gin.H{"error": "Script file not found: " + scriptPath}) + return + } + + // 6. 防抖 + 尾部补刀 + lockKey := fmt.Sprintf("%s:%s", repoName, ref) + now := time.Now() + + entryVal, _ := pendingDebounceMap.LoadOrStore(lockKey, &debounceEntry{ + scriptPath: scriptPath, + repoName: repoName, + ref: ref, + }) + entry := entryVal.(*debounceEntry) + + entry.mu.Lock() + + // 是否首次/窗口外执行:没有活动定时器即视为窗口外 + // (timer==nil 涵盖首次 lastExec==0 和上一次 timer fire 后两种情况) + firstRun := entry.timer == nil + + if firstRun { + // 立即执行本次请求的 commitID;同时安排尾部补刀捕获期间新 push + entry.lastExec = now + entry.pendingCommit = commitID // 兜底:timer fire 时无新 push 则用本次 commitID + entry.scheduleLocked(lockKey) + entry.mu.Unlock() + + log.Printf("[INFO] 开始执行脚本: %s (仓库: %s, 分支: %s, commit: %s)\n", + scriptPath, repoName, ref, shortCommit(commitID)) + go runScript(scriptPath, repoName, ref, commitID) + + c.JSON(http.StatusOK, gin.H{ + "status": "triggered", + "repository": repoName, + "ref": ref, + "script": scriptPath, + }) + return + } + + // 防抖窗口内(活动定时器存在):续命定时器,使用最新 commitID + remainingSeconds := max(int((debounceDuration - now.Sub(entry.lastExec)).Seconds()), 0) + entry.pendingCommit = commitID + entry.scheduleLocked(lockKey) + entry.mu.Unlock() + + msg := fmt.Sprintf("触发太频繁,防抖拦截中(5分钟限制),剩余等待时间: %d 秒", remainingSeconds) + log.Printf("[DEBOUNCE] 跳过执行 [%s] -> %s (已安排尾部补刀)\n", lockKey, msg) + c.JSON(http.StatusOK, gin.H{ + "status": "debounced", + "message": msg, + "remaining_seconds": remainingSeconds, + }) +}