package main import ( "bytes" "context" "encoding/json" "fmt" "log" "net/http" "net/url" "time" commitcontract "git.cin.su/m/Assistant/contracts/commit" "git.cin.su/m/Assistant/contracts/continuity" "git.cin.su/m/Assistant/platform/ids" ) type dueReminderRepository interface { DueArmed(context.Context, int64, int) ([]continuity.ProspectiveItem, error) ClaimDueReminder(context.Context, continuity.ProspectiveItemID, continuity.ProspectiveRevision, continuity.AttentionItemID, int64) (continuity.ProspectiveItem, continuity.AttentionItem, error) PendingAttention(context.Context, int) ([]continuity.AttentionItem, error) MarkAttentionDelivering(context.Context, continuity.AttentionItemID, continuity.AttentionRevision) (continuity.AttentionItem, error) MarkAttentionDelivered(context.Context, continuity.AttentionItemID, continuity.AttentionRevision, continuity.CommittedActionIntentID, continuity.CommittedAuthorizedActionID, continuity.CommittedActionAttemptID, string, int64) (continuity.AttentionItem, error) MarkAttentionFailed(context.Context, continuity.AttentionItemID, continuity.AttentionRevision, continuity.CommittedActionIntentID, continuity.CommittedAuthorizedActionID, continuity.CommittedActionAttemptID, string) (continuity.AttentionItem, error) Load(context.Context, continuity.ProspectiveItemID) (continuity.ProspectiveItem, error) } type dueNotificationClient struct { baseURL *url.URL token string client *http.Client } type dueNotifier interface { NotifyDue(context.Context, continuity.ProspectiveItem) (dueNotificationResponse, error) } type dueNotificationRequest struct { ProspectiveItemID continuity.ProspectiveItemID `json:"prospective_item_id"` ExpectedRevision continuity.ProspectiveRevision `json:"expected_revision"` } type dueNotificationResponse struct { AuthorizedActionID commitcontract.AuthorizedActionID `json:"AuthorizedActionID"` ActionIntentID commitcontract.ActionIntentID `json:"ActionIntentID"` ActionAttemptID commitcontract.ActionAttemptID `json:"ActionAttemptID"` State commitcontract.ActionState `json:"State"` } func newDueNotificationClient(rawURL, token string, client *http.Client) (dueNotificationClient, error) { base, err := url.Parse(rawURL) if err != nil || base.Scheme == "" || base.Host == "" || token == "" { return dueNotificationClient{}, fmt.Errorf("invalid Commit due notification configuration") } if client == nil { client = http.DefaultClient } return dueNotificationClient{baseURL: base, token: token, client: client}, nil } func (c dueNotificationClient) NotifyDue(ctx context.Context, item continuity.ProspectiveItem) (dueNotificationResponse, error) { target := *c.baseURL target.Path = "/v1/internal/continuity/notify-due" payload, err := json.Marshal(dueNotificationRequest{ProspectiveItemID: item.ID, ExpectedRevision: item.Revision}) if err != nil { return dueNotificationResponse{}, err } request, err := http.NewRequestWithContext(ctx, http.MethodPost, target.String(), bytes.NewReader(payload)) if err != nil { return dueNotificationResponse{}, err } request.Header.Set("Authorization", "Bearer "+c.token) request.Header.Set("Content-Type", "application/json") response, err := c.client.Do(request) if err != nil { return dueNotificationResponse{}, err } defer response.Body.Close() if response.StatusCode != http.StatusCreated && response.StatusCode != http.StatusAccepted { return dueNotificationResponse{}, fmt.Errorf("commit due notification status %d", response.StatusCode) } var result dueNotificationResponse if err := json.NewDecoder(response.Body).Decode(&result); err != nil { return dueNotificationResponse{}, err } if result.ActionIntentID == "" || result.AuthorizedActionID == "" || result.ActionAttemptID == "" { return dueNotificationResponse{}, fmt.Errorf("invalid commit due notification response") } return result, nil } type dueReminderWorker struct { repository dueReminderRepository commit dueNotifier now func() time.Time newID func() (ids.ID, error) interval time.Duration limit int } func (w dueReminderWorker) Start(ctx context.Context) { if w.interval <= 0 { w.interval = 15 * time.Second } if w.limit <= 0 { w.limit = 10 } go func() { w.process(ctx) ticker := time.NewTicker(w.interval) defer ticker.Stop() for { select { case <-ctx.Done(): return case <-ticker.C: w.process(ctx) } } }() } func (w dueReminderWorker) process(ctx context.Context) { now := w.now() if err := w.claimDue(ctx, now); err != nil { log.Printf("due reminder claim skipped: %v", err) } if err := w.deliverPending(ctx, now); err != nil { log.Printf("due reminder delivery skipped: %v", err) } } func (w dueReminderWorker) claimDue(ctx context.Context, now time.Time) error { items, err := w.repository.DueArmed(ctx, now.UnixMilli(), w.limit) if err != nil { return err } for _, item := range items { id, err := w.newID() if err != nil { return err } if _, _, err = w.repository.ClaimDueReminder(ctx, item.ID, item.Revision, continuity.AttentionItemID(id), now.UnixMilli()); err != nil { log.Printf("due reminder %s not claimed: %v", item.ID, err) } } return nil } func (w dueReminderWorker) deliverPending(ctx context.Context, now time.Time) error { items, err := w.repository.PendingAttention(ctx, w.limit) if err != nil { return err } for _, attention := range items { if attention.SourceKind != continuity.AttentionSourceProspectiveReminder || attention.SourceProspectiveID == "" { continue } delivering, err := w.repository.MarkAttentionDelivering(ctx, attention.ID, attention.Revision) if err != nil { log.Printf("attention %s not marked delivering: %v", attention.ID, err) continue } prospective, err := w.repository.Load(ctx, delivering.SourceProspectiveID) if err != nil { log.Printf("attention %s source unavailable: %v", attention.ID, err) continue } result, err := w.commit.NotifyDue(ctx, prospective) if err != nil { log.Printf("attention %s notification unavailable: %v", attention.ID, err) continue } if result.State != commitcontract.ActionSucceeded { _, err = w.repository.MarkAttentionFailed(ctx, delivering.ID, delivering.Revision, continuity.CommittedActionIntentID(result.ActionIntentID), continuity.CommittedAuthorizedActionID(result.AuthorizedActionID), continuity.CommittedActionAttemptID(result.ActionAttemptID), string(result.State)) if err != nil { log.Printf("attention %s delivery failure outcome not persisted: %v", attention.ID, err) } continue } _, err = w.repository.MarkAttentionDelivered(ctx, delivering.ID, delivering.Revision, continuity.CommittedActionIntentID(result.ActionIntentID), continuity.CommittedAuthorizedActionID(result.AuthorizedActionID), continuity.CommittedActionAttemptID(result.ActionAttemptID), string(result.State), now.UnixMilli()) if err != nil { log.Printf("attention %s delivery outcome not persisted: %v", attention.ID, err) } } return nil }