// Copyright 2026 Alibaba Group // Licensed under the Apache License, Version 2.0 (the "License"); // you may not use this file except in compliance with the License. // You may obtain a copy of the License at // // http://www.apache.org/licenses/LICENSE-2.0 // // Unless required by applicable law or agreed to in writing, software // distributed under the License is distributed on an "AS IS" BASIS, // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. // See the License for the specific language governing permissions and // limitations under the License. package personal import ( "encoding/json" "errors" "fmt" "os" "path/filepath" "sort" "time" eventlock "github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event/lock" "github.com/DingTalk-Real-AI/dingtalk-workspace-cli/pkg/config" ) const ( StateFileName = "personal_subscriptions.json" StateLockFileName = "personal_subscriptions.lock" stateLockWaitTimeout = 5 * time.Second stateLockRetryDelay = 25 * time.Millisecond ) var ( runStateReadFile = os.ReadFile runStateMkdirAll = os.MkdirAll runStateTryAcquire = eventlock.TryAcquire runStateRemove = os.Remove runStateMarshalIndent = json.MarshalIndent runStateWriteFile = os.WriteFile runStateRename = os.Rename runStateSleep = time.Sleep ) type RunState struct { SubscribeID string `json:"subscribe_id"` EventKey string `json:"event_key,omitempty"` RuleType string `json:"rule_type,omitempty"` ClientID string `json:"client_id,omitempty"` SourceID string `json:"source_id,omitempty"` IdentityHash string `json:"identity_hash,omitempty"` CreatedAt time.Time `json:"created_at"` } func LoadRunStates(workDir string) ([]RunState, error) { path := filepath.Join(workDir, StateFileName) b, err := runStateReadFile(path) if os.IsNotExist(err) { return nil, nil } if err != nil { return nil, err } var states []RunState if err := json.Unmarshal(b, &states); err != nil { return nil, err } sort.Slice(states, func(i, j int) bool { return states[i].SubscribeID < states[j].SubscribeID }) return states, nil } func UpsertRunState(workDir string, st RunState) error { if st.SubscribeID == "" { return nil } if st.CreatedAt.IsZero() { st.CreatedAt = time.Now().UTC() } return withRunStateLock(workDir, stateLockWaitTimeout, func() error { states, err := LoadRunStates(workDir) if err != nil { return err } replaced := false for i := range states { if states[i].SubscribeID == st.SubscribeID { states[i] = st replaced = true break } } if !replaced { states = append(states, st) } return writeRunStates(workDir, states) }) } func RemoveRunStates(workDir string, subscribeIDs []string) error { if len(subscribeIDs) == 0 { return nil } remove := make(map[string]struct{}, len(subscribeIDs)) for _, id := range subscribeIDs { if id != "" { remove[id] = struct{}{} } } return withRunStateLock(workDir, stateLockWaitTimeout, func() error { states, err := LoadRunStates(workDir) if err != nil { return err } filtered := states[:0] for _, st := range states { if _, ok := remove[st.SubscribeID]; !ok { filtered = append(filtered, st) } } return writeRunStates(workDir, filtered) }) } func withRunStateLock(workDir string, wait time.Duration, fn func() error) error { if err := runStateMkdirAll(workDir, config.DirPerm); err != nil { return err } lockPath := filepath.Join(workDir, StateLockFileName) deadline := time.Now().Add(wait) for { held, err := runStateTryAcquire(lockPath) if err == nil { defer held.Close() return fn() } if !errors.Is(err, eventlock.ErrBusy) { return err } remaining := time.Until(deadline) if remaining <= 0 { return fmt.Errorf("personal event: timed out waiting for run-state lock after %s", wait) } delay := stateLockRetryDelay if remaining < delay { delay = remaining } runStateSleep(delay) } } func writeRunStates(workDir string, states []RunState) error { if err := runStateMkdirAll(workDir, config.DirPerm); err != nil { return err } sort.Slice(states, func(i, j int) bool { return states[i].SubscribeID < states[j].SubscribeID }) path := filepath.Join(workDir, StateFileName) if len(states) == 0 { if err := runStateRemove(path); err != nil && !os.IsNotExist(err) { return err } return nil } b, err := runStateMarshalIndent(states, "", " ") if err != nil { return err } b = append(b, '\n') tmp := path + ".tmp" if err := runStateWriteFile(tmp, b, config.FilePerm); err != nil { return err } if err := runStateRename(tmp, path); err != nil { _ = runStateRemove(tmp) return err } return nil }