vend with

2026-06-27 16:44:47 -06:00
parent 48c3474c15
commit a9f78655c3
20 changed files with 1068 additions and 3 deletions
@@ -0,0 +1,75 @@
+package dbos
+
+import (
+	"context"
+	"fmt"
+	"log"
+	"reflect"
+	"runtime"
+	"strings"
+	"time"
+
+	"github.com/dbos-inc/dbos-transact-golang/dbos"
+)
+
+type Client struct {
+	c dbos.Client
+	q string
+}
+
+func QueueClient(ctx context.Context, conn, q string, foo func(*Client) error) error {
+	connctx, conncan := context.WithTimeout(ctx, 10*time.Second)
+	defer conncan()
+	c, err := dbos.NewClient(connctx, dbos.ClientConfig{
+		DatabaseURL: conn,
+	})
+	if err != nil {
+		return err
+	}
+	defer c.Shutdown(5 * time.Second)
+	return foo(&Client{c: c, q: q})
+}
+
+func (c *Client) WithQueue(q string) *Client {
+	c2 := *c
+	c2.q = q
+	return &c2
+}
+
+func Go[P any, R any](ctx context.Context, c *Client, foo dbos.Workflow[P, R], input P, globalDedupe, concurrentDedupe string) error {
+	log.Printf("[client] going to %s#%s#%s", fooToName(foo), globalDedupe, concurrentDedupe)
+
+	result := make(chan error)
+	go func() {
+		defer close(result)
+		_, err := dbos.Enqueue[P, R](c.c, c.q, fooToName(foo), input, dbos.WithEnqueueDeduplicationID(concurrentDedupe), dbos.WithEnqueueApplicationVersion("latest"), dbos.WithEnqueueWorkflowID(globalDedupe))
+		select {
+		case <-ctx.Done():
+		case result <- err:
+		}
+	}()
+	select {
+	case err := <-result:
+		return err
+	case <-ctx.Done():
+	}
+	return ctx.Err()
+}
+
+// https://github.com/dbos-inc/dbos-transact-golang/blob/0d755bda26e3162c823f3e316aa0f88677b8b246/dbos/workflow.go#L577
+func fooToName[P any, R any](fn dbos.Workflow[P, R]) string {
+	ptr := reflect.ValueOf(fn).Pointer()
+	fqn := runtime.FuncForPC(ptr).Name()
+
+	// If this is a generic workflow, append the actual types to the FQN
+	if strings.Contains(fqn, "[") {
+		fqn = strings.Split(fqn, "[")[0]
+		fqn = fmt.Sprintf("%s[%s,%s]",
+			fqn,
+			reflect.TypeFor[P]().String(),
+			reflect.TypeFor[R]().String(),
+		)
+	}
+
+	return fqn
+}
@@ -0,0 +1,122 @@
+package dbos
+
+import (
+	"context"
+	"os"
+	"regexp"
+	"runtime"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/dbos-inc/dbos-transact-golang/dbos"
+)
+
+func TestDBOS(t *testing.T) {
+	conn := os.Getenv("CONN_URL")
+	if conn == "" {
+		t.Skip("no $CONN_URL")
+	}
+
+	routine := func() string {
+		buf := make([]byte, 128_000)
+		n := runtime.Stack(buf[:], false)
+		buf = buf[:n]
+
+		myRoutine := strings.Fields(strings.TrimPrefix(string(buf), "goroutine"))[0]
+
+		re := regexp.MustCompile(`created by .*? goroutine [0-9]+`)
+		chain := []string{myRoutine}
+		for _, match := range re.FindAllString(string(buf), -1) {
+			fields := strings.Fields(match)
+			chain = append(chain, fields[len(fields)-1])
+		}
+		return strings.Join(chain, "-")
+	}
+
+	ctx, can := context.WithTimeout(context.Background(), 19*time.Second)
+	defer can()
+
+	wg := &sync.WaitGroup{}
+	defer wg.Wait()
+	wg.Add(3)
+	waitScheduled := &sync.Once{}
+
+	onDemandWorkflow := func(ctx dbos.DBOSContext, arg string) (string, error) {
+		t.Logf("[%s] on demand...", routine())
+		defer t.Logf("[%s] /on demand", routine())
+
+		dbos.Sleep(ctx, 1*time.Millisecond)
+		return dbos.RunAsStep(ctx, func(ctx context.Context) (string, error) {
+			t.Logf("[%s] on demand step...", routine())
+			defer t.Logf("[%s] /on demand step", routine())
+
+			t.Logf("[%s] step on demand", routine())
+			wg.Done()
+			return "my step string", nil
+		})
+	}
+
+	scheduledWorkflow := func(dbos.DBOSContext, time.Time) (int, error) {
+		t.Logf("[%s] scheduled...", routine())
+		defer t.Logf("[%s] /scheduled", routine())
+
+		waitScheduled.Do(wg.Done)
+		return -1, nil
+	}
+
+	t.Run("worker", func(t *testing.T) {
+		go func() {
+			if err := QueueWorker(ctx, conn, "queue", func(w *Worker) error {
+				t.Logf("[%s] dbosc...", routine())
+				defer t.Logf("[%s] /dbosc", routine())
+
+				t.Logf("[%s] dbosc.can...", routine())
+				Can(w, onDemandWorkflow)
+				t.Logf("[%s] dbosc.every...", routine())
+				Every(w, scheduledWorkflow, "* * * * * *")
+				t.Logf("[%s] dbosc go do...", routine())
+				go func() {
+					t.Logf("[%s] dbos do...", routine())
+					defer t.Logf("[%s] /dbos do", routine())
+
+					time.Sleep(2 * time.Second)
+
+					res, err := Do(ctx, w, onDemandWorkflow, "arg")
+					if err != nil {
+						t.Fatal(err)
+					}
+					t.Logf("onDemandWorkflow result: %q", res)
+				}()
+
+				t.Logf("[%s] dbosc listen...", routine())
+				defer t.Logf("[%s] /dbosc listen", routine())
+				return w.Listen(ctx)
+			}); err != nil {
+				t.Fatal(err)
+			}
+		}()
+	})
+
+	t.Run("client", func(t *testing.T) {
+		if err := QueueClient(ctx, conn, "queue", func(w *Client) error {
+			t.Logf("[%s] dbosc...", routine())
+			defer t.Logf("[%s] /dbosc", routine())
+
+			t.Logf("[%s] dbos go...", routine())
+			defer t.Logf("[%s] /dbos go", routine())
+
+			err := Go(ctx, w, onDemandWorkflow, "arg", time.Now().String())
+			if err != nil {
+				t.Fatal(err)
+			}
+			t.Logf("[%s] onDemandWorkflow enqueued result", routine())
+
+			return nil
+		}); err != nil {
+			t.Fatal(err)
+		}
+	})
+
+}
@@ -0,0 +1,103 @@
+package dbos
+
+import (
+	"context"
+	"log"
+	"os"
+	"slices"
+	"time"
+
+	"github.com/dbos-inc/dbos-transact-golang/dbos"
+)
+
+type Worker struct {
+	dbos dbos.DBOSContext
+	q    string
+	qs   []dbos.WorkflowQueue
+}
+
+func QueueWorker(ctx context.Context, conn, q string, foo func(*Worker) error) error {
+	return NewWorker(ctx, conn, func(w *Worker) error {
+		return foo(w.WithQueue(q))
+	})
+}
+
+func NewWorker(ctx context.Context, conn string, foo func(*Worker) error) error {
+	type result struct {
+		ctx dbos.DBOSContext
+		err error
+	}
+	ch := make(chan result)
+	go func() {
+		defer close(ch)
+		dbosctx, err := dbos.NewDBOSContext(ctx, dbos.Config{
+			AppName:            "with",
+			DatabaseURL:        conn,
+			AdminServer:        os.Getenv("WITH_DBOS_ADMIN_SERVER") == "true",
+			ApplicationVersion: "latest",
+		})
+		select {
+		case ch <- result{ctx: dbosctx, err: err}:
+		case <-ctx.Done():
+		}
+	}()
+	select {
+	case result := <-ch:
+		if err := result.err; err != nil {
+			return err
+		}
+		return foo(&Worker{dbos: result.ctx})
+	case <-ctx.Done():
+	}
+	return ctx.Err()
+}
+
+func (c *Worker) Listen(ctx context.Context) error {
+	dbos := c.dbos
+	//dbos.ListenQueues(c.dbos, c.qs...)
+	log.Printf("[worker] listening to %+v", c.q)
+	if err := dbos.Launch(); err != nil {
+		return err
+	}
+	defer dbos.Shutdown(5 * time.Second)
+	<-ctx.Done()
+	return nil
+}
+
+func (c *Worker) WithQueue(q string) *Worker {
+	c2 := *c
+	c2.q = q
+	c2.qs = slices.Clone(c2.qs)
+
+	dbosq := dbos.NewWorkflowQueue(c2.dbos, q)
+	//dbos.WithWorkerConcurrency(5),  // per worker
+	//dbos.WithGlobalConcurrency(10), // per all workers
+	//dbos.WithRateLimiter(&dbos.RateLimiter{Limit: 100, Period: time.Second}),
+	//dbos.WithPriorityEnabled(),
+	// dbos.WithPartitionQueue(), // WithQueuePartitionKey but has limit per-partiiton
+
+	c2.qs = append(c2.qs, dbosq)
+
+	return &c2
+}
+
+func Can[P any, R any](c *Worker, foo dbos.Workflow[P, R]) {
+	log.Printf("[worker] registering %s", fooToName(foo))
+	dbos.RegisterWorkflow(c.dbos, foo, dbos.WithWorkflowName(fooToName(foo)))
+}
+
+func Every[P any, R any](w *Worker, foo dbos.Workflow[P, R], cron string) {
+	log.Printf("[worker] registering cron '%s' %s", cron, fooToName(foo))
+	dbos.RegisterWorkflow(w.dbos, foo, dbos.WithSchedule(cron), dbos.WithWorkflowName(fooToName(foo)))
+}
+
+func Do[P any, R any](ctx context.Context, w *Worker, foo dbos.Workflow[P, R], input P) (R, error) {
+	dbosctx := dbos.From(w.dbos, ctx)
+	log.Printf("[worker] doing %s", fooToName(foo))
+	handle, err := dbos.RunWorkflow(dbosctx, foo, input) //, options...)
+	var some R
+	if err != nil {
+		return some, err
+	}
+	return handle.GetResult()
+}