pkg/server

package main

import (
	"context"
	"fmt"
	"os"
	"os/signal"
	"syscall"

	"github.com/dexadata/dexaflow/pkg/server"
)

func main() {
	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
	err := server.Run(ctx)
	stop()
	if err != nil {
		fmt.Fprintln(os.Stderr, err)
		os.Exit(server.ExitCode(err))
	}
}

package main

import (
	"context"
	"fmt"
	"os"
	"os/signal"
	"syscall"

	"github.com/dexadata/dexaflow/pkg/admission"
	"github.com/dexadata/dexaflow/pkg/server"
)

func main() {
	freeze := admission.HookFunc(func(_ context.Context, req *admission.Request) (admission.Decision, error) {
		if req.TenantID == "acme" {
			return admission.Refuse("acme is frozen until 06:00 UTC"), nil
		}
		return admission.Admit(), nil
	})
	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
	err := server.Run(ctx, server.WithAdmission(freeze))
	stop()
	if err != nil {
		fmt.Fprintln(os.Stderr, err)
		os.Exit(server.ExitCode(err))
	}
}

package main

import (
	"context"
	"fmt"
	"os"
	"os/signal"
	"strings"
	"syscall"

	"github.com/dexadata/dexaflow/pkg/authn"
	"github.com/dexadata/dexaflow/pkg/server"
)

// directory signs users in against a company directory, here a fixed table,
// and recognizes the tokens it hands out. tenantID is the tenant's name.
type directory struct {
	tenantID string
	users    map[string]string
}

func (d directory) IssueToken(_ context.Context, c authn.Credentials) (string, error) {
	if pw, ok := d.users[c.Username]; !ok || pw != c.Password {
		return "", authn.ErrInvalidCredentials
	}
	return "dir:" + c.Username, nil
}

func (d directory) Authenticate(_ context.Context, token string) (*authn.User, error) {
	email, ok := strings.CutPrefix(token, "dir:")
	if !ok || d.users[email] == "" {
		return nil, authn.ErrInvalidToken
	}
	return &authn.User{
		ID: email, TenantID: d.tenantID, Email: email,
		Roles:       []string{"viewer"},
		Permissions: []authn.Permission{{Action: "read", Resource: "*"}},
	}, nil
}

func main() {
	dir := directory{tenantID: "default", users: map[string]string{"ada@example.com": "s3cret"}}
	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
	err := server.Run(ctx, server.WithAuthenticator(dir))
	stop()
	if err != nil {
		fmt.Fprintln(os.Stderr, err)
		os.Exit(server.ExitCode(err))
	}
}

package main

import (
	"context"
	"fmt"
	"os"
	"os/signal"
	"syscall"

	"github.com/dexadata/dexaflow/pkg/server"
)

func main() {
	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
	err := server.Run(ctx, server.WithConfigFile("/etc/acme/dexaflow-server.yaml"))
	stop()
	if err != nil {
		fmt.Fprintln(os.Stderr, err)
		os.Exit(server.ExitCode(err))
	}
}

package main

import (
	"context"
	"encoding/json"
	"fmt"
	"io"
	"net/url"
	"os"
	"os/signal"
	"path/filepath"
	"strconv"
	"strings"
	"syscall"

	"github.com/dexadata/dexaflow/pkg/logsink"
	"github.com/dexadata/dexaflow/pkg/server"
)

// fileSink stores each attempt's log as one file under dir, its name built
// from the Ref with every field escaped, so no value can leave dir.
type fileSink struct{ dir string }

func (s fileSink) path(ref logsink.Ref) string {
	name := strings.Join([]string{
		url.PathEscape(ref.TenantID), url.PathEscape(ref.DagID), url.PathEscape(ref.RunID),
		url.PathEscape(ref.TaskID), strconv.Itoa(ref.TryNumber), strconv.Itoa(ref.AttemptEpoch),
	}, "_")
	return filepath.Join(s.dir, name+".jsonl")
}

func (s fileSink) Open(ref logsink.Ref) (logsink.LogWriter, error) {
	f, err := os.Create(s.path(ref))
	if err != nil {
		return nil, err
	}
	return &fileWriter{f: f, enc: json.NewEncoder(f)}, nil
}

// Read returns the stored file; os.Open wraps fs.ErrNotExist for a missing
// one, which the API reports as no log yet.
func (s fileSink) Read(_ context.Context, ref logsink.Ref) (io.ReadCloser, error) {
	return os.Open(s.path(ref))
}

type fileWriter struct {
	f   *os.File
	enc *json.Encoder
}

func (w *fileWriter) WriteEvent(ev logsink.Event) error { return w.enc.Encode(ev) }
func (w *fileWriter) Close() error                      { return w.f.Close() }

func main() {
	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
	err := server.Run(ctx, server.WithLogSink(fileSink{dir: "/var/lib/task-logs"}))
	stop()
	if err != nil {
		fmt.Fprintln(os.Stderr, err)
		os.Exit(server.ExitCode(err))
	}
}

package main

import (
	"context"
	"errors"
	"fmt"
	"os"
	"os/signal"
	"syscall"

	"github.com/dexadata/dexaflow/pkg/authn"
	"github.com/dexadata/dexaflow/pkg/server"
)

// portal verifies the handoff tokens of the operator's own portal, here a
// fixed table of one-time codes.
type portal struct{ codes map[string]authn.Identity }

func (portal) Provider() string { return "portal" }

func (p portal) Verify(_ context.Context, raw string) (*authn.Identity, error) {
	id, ok := p.codes[raw]
	if !ok {
		return nil, errors.New("unknown handoff code")
	}
	return &id, nil
}

func main() {
	p := portal{codes: map[string]authn.Identity{"c-1": {Subject: "u-42", Email: "ada@example.com", Tenant: "default"}}}
	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
	err := server.Run(ctx, server.WithTrustedIssuer(p, "https://portal.example.com"))
	stop()
	if err != nil {
		fmt.Fprintln(os.Stderr, err)
		os.Exit(server.ExitCode(err))
	}
}

package main

import (
	"context"
	"fmt"
	"io/fs"
	"net/http"
	"os"
	"os/signal"
	"syscall"

	"github.com/dexadata/dexaflow/pkg/server"
)

// shellUI serves a UI built into dist/ by its own toolchain: the bundle under
// /static/ and the same index.html for every client side route.
type shellUI struct{ dist fs.FS }

func (u shellUI) StaticHandler() http.Handler { return http.FileServerFS(u.dist) }

func (u shellUI) Index(w http.ResponseWriter, _ string) {
	b, err := fs.ReadFile(u.dist, "index.html")
	if err != nil {
		http.Error(w, "the UI is not built", http.StatusInternalServerError)
		return
	}
	w.Header().Set("Content-Type", "text/html; charset=utf-8")
	_, _ = w.Write(b)
}

func main() {
	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
	err := server.Run(ctx, server.WithUIServer(shellUI{dist: os.DirFS("dist")}))
	stop()
	if err != nil {
		fmt.Fprintln(os.Stderr, err)
		os.Exit(server.ExitCode(err))
	}
}

package main

import (
	"context"
	"fmt"
	"os"
	"os/signal"
	"path"
	"sync"
	"syscall"
	"time"

	"github.com/dexadata/dexaflow/pkg/server"
	"github.com/dexadata/dexaflow/pkg/xcomstore"
)

// mapBackend keeps XCom values in the server's memory: enough for a single
// replica that may lose them on a restart, as a test install can.
type mapBackend struct {
	mu sync.Mutex
	m  map[string]mapEntry
}

type mapEntry struct {
	e       xcomstore.Entry
	expires time.Time
}

func (b *mapBackend) Push(_ context.Context, key string, e xcomstore.Entry, ttl time.Duration) error {
	b.mu.Lock()
	defer b.mu.Unlock()
	b.m[key] = mapEntry{e: e, expires: time.Now().Add(ttl)}
	return nil
}

func (b *mapBackend) Fetch(_ context.Context, key string) (xcomstore.Entry, error) {
	b.mu.Lock()
	defer b.mu.Unlock()
	v, ok := b.m[key]
	if !ok || time.Now().After(v.expires) {
		return xcomstore.Entry{}, xcomstore.ErrNotFound
	}
	return v.e, nil
}

func (b *mapBackend) Delete(_ context.Context, key string) error {
	b.mu.Lock()
	defer b.mu.Unlock()
	delete(b.m, key)
	return nil
}

func (b *mapBackend) List(_ context.Context, pattern string) ([]string, error) {
	b.mu.Lock()
	defer b.mu.Unlock()
	var keys []string
	for k, v := range b.m {
		if ok, _ := path.Match(pattern, k); ok && time.Now().Before(v.expires) {
			keys = append(keys, k)
		}
	}
	return keys, nil
}

func main() {
	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
	err := server.Run(ctx, server.WithXComBackend(&mapBackend{m: map[string]mapEntry{}}))
	stop()
	if err != nil {
		fmt.Fprintln(os.Stderr, err)
		os.Exit(server.ExitCode(err))
	}
}