Assistant/cmd/continuity-runtime/due_worker.go

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
}