orbit/examples/result-inbox/cmd/receiver/main.go

// receiver 使用独立业务数据库演示结果Inbox与版本化业务视图。
package main

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"os"
	"os/signal"
	"syscall"
	"time"

	"git.yuntuo.hk/k2safe/orbit/examples/result-inbox/internal/inbox"
	orbit "git.yuntuo.hk/k2safe/orbit/sdk/go"
)

func main() {
	if err := run(); err != nil {
		fmt.Fprintln(os.Stderr, err)
		os.Exit(1)
	}
}
func run() error {
	if len(os.Args) != 2 || (os.Args[1] != "init" && os.Args[1] != "consume" && os.Args[1] != "demo") {
		return errors.New("usage: receiver init|consume|demo")
	}
	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
	defer stop()
	open, cancel := context.WithTimeout(ctx, 10*time.Second)
	db, err := inbox.Open(open, os.Getenv("ORBIT_RECEIVER_DSN"))
	cancel()
	if err != nil {
		return err
	}
	defer db.Close()
	if os.Args[1] == "init" {
		bounded, stop := context.WithTimeout(ctx, 10*time.Second)
		defer stop()
		if err = inbox.Init(bounded, db); err != nil {
			return err
		}
		fmt.Println("receiver_schema_initialized=true foreign_keys=0")
		return nil
	}
	api, queue, err := client()
	if err != nil {
		return err
	}
	name := os.Getenv("RECEIVER_NAME")
	if name == "" {
		name = "task-result-view"
	}
	store, err := inbox.New(db, name, queue)
	if err != nil {
		return err
	}
	if os.Args[1] == "demo" {
		return demo(ctx, api, store, queue)
	}
	check, checkStop := context.WithTimeout(ctx, 5*time.Second)
	subscription, err := api.GetSubscription(check, os.Getenv("ORBIT_SUBSCRIPTION_ID"))
	checkStop()
	if err != nil {
		return err
	}
	if subscription.QueueID != queue {
		return errors.New("subscription does not match configured receiver queue")
	}
	consumer, err := orbit.NewEventConsumer(api, orbit.EventConsumerOptions{SubscriptionID: subscription.ID, Capacity: 4, Handle: func(work context.Context, event orbit.ResultEvent) error {
		err := store.Handle(work, event)
		if err != nil {
			fmt.Fprintf(os.Stderr, "receiver_event=%s error=%v\n", event.ID, err)
		}
		return err
	}, ShutdownTimeout: 10 * time.Second})
	if err != nil {
		return err
	}
	observed := make(chan struct{})
	go func() {
		defer close(observed)
		for event := range consumer.Events() {
			fmt.Printf("consumer_event=%s event_id=%s code=%s\n", event.Type, event.EventID, event.Code)
		}
	}()
	err = consumer.Run(ctx)
	<-observed
	return err
}
func client() (*orbit.Client, string, error) {
	token, queue := os.Getenv("ORBIT_TOKEN"), os.Getenv("ORBIT_QUEUE_ID")
	if path := os.Getenv("ORBIT_CREDENTIALS_FILE"); path != "" {
		raw, err := os.ReadFile(path)
		if err != nil {
			return nil, "", errors.New("cannot read credential file")
		}
		var c struct {
			Token   string `json:"token"`
			QueueID string `json:"queue_id"`
		}
		if json.Unmarshal(raw, &c) != nil {
			return nil, "", errors.New("invalid credential file")
		}
		if token == "" {
			token = c.Token
		}
		if queue == "" {
			queue = c.QueueID
		}
	}
	if token == "" || queue == "" {
		return nil, "", errors.New("event-enabled credential and queue are required")
	}
	c, err := orbit.NewClient(os.Getenv("ORBIT_URL"), orbit.Options{Token: token})
	return c, queue, err
}
func demo(parent context.Context, api *orbit.Client, store *inbox.Store, queue string) error {
	ctx, stop := context.WithTimeout(parent, 20*time.Second)
	defer stop()
	id, err := orbit.NewEventMessageID()
	if err != nil {
		return err
	}
	created, err := api.CreateSubscription(ctx, "inbox-sub-"+id, orbit.CreateSubscriptionRequest{Name: "inbox-" + id, QueueID: queue, Types: []string{"task.cancelled"}, MaxPending: 10, MaxAttempts: 3})
	if err != nil {
		return err
	}
	task, err := api.SubmitTask(ctx, "inbox-task-"+id, orbit.SubmitTaskRequest{QueueID: queue, Handler: "object.prepare", HandlerVersion: "v1", PayloadVersion: "v1", Payload: json.RawMessage(`{}`)})
	if err != nil {
		return err
	}
	if _, err = api.CancelTask(ctx, task.Task.ID, "inbox-cancel-"+id, orbit.CancelTaskRequest{ExpectedRevision: task.Task.Revision, Reason: "durable inbox example"}); err != nil {
		return err
	}
	duplicate := make(chan bool, 1)
	consumer, err := orbit.NewEventConsumer(api, orbit.EventConsumerOptions{SubscriptionID: created.Subscription.ID, Capacity: 1, ShutdownTimeout: 5 * time.Second, Handle: func(work context.Context, event orbit.ResultEvent) error {
		if event.Snapshot.TaskID != task.Task.ID {
			return inbox.ErrInvalidEvent
		}
		if _, err := store.Apply(work, event); err != nil {
			return err
		}
		// 再用同一冻结事件进入业务事务，证明重复不会再次推进视图；真实进程故障另有回归。
		again, err := store.Apply(work, event)
		if err != nil {
			return err
		}
		if !again.Duplicate {
			return errors.New("receiver duplicate was not recognized")
		}
		select {
		case duplicate <- true:
		default:
		}
		return nil
	}})
	if err != nil {
		return err
	}
	run, cancel := context.WithCancel(ctx)
	defer cancel()
	done := make(chan error, 1)
	go func() { done <- consumer.Run(run) }()
	confirmed := false
	for !confirmed {
		select {
		case <-ctx.Done():
			cancel()
			<-done
			return errors.New("inbox demo timed out before confirmed acknowledgement")
		case err := <-done:
			if err == nil {
				err = errors.New("consumer stopped before acknowledgement")
			}
			return err
		case event, ok := <-consumer.Events():
			if !ok {
				if err := <-done; err != nil {
					return err
				}
				return errors.New("consumer stopped before acknowledgement")
			}
			confirmed = event.Type == "acknowledged"
		}
	}
	cancel()
	if err = <-done; err != nil {
		return err
	}
	select {
	case <-duplicate:
	default:
		return errors.New("duplicate verification missing")
	}
	view, err := store.View(ctx, task.Task.ID)
	if err != nil {
		return err
	}
	if view.AppliedEvents != 1 || view.LastState != "CANCELLED" {
		return errors.New("duplicate changed business projection")
	}
	paused, err := api.ConfigureSubscription(ctx, created.Subscription.ID, "inbox-pause-"+id, orbit.ConfigureSubscriptionRequest{ExpectedRevision: created.Subscription.Revision, Enabled: false, Reason: "durable inbox example completed"})
	if err != nil {
		return err
	}
	fmt.Printf("task=%s state=%s inbox_duplicate=true applied_events=%d\n", task.Task.ID, view.LastState, view.AppliedEvents)
	fmt.Printf("ack_confirmed=%d pending=%d subscription_enabled=%t\n", consumer.Stats().Acknowledged, paused.Subscription.PendingCount, paused.Subscription.Enabled)
	return nil
}