johnandersen777.bsky.social/socialweb-computer-kcp

Switch repository

Loading…

package plc

import (
	"cmp"
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"net/http"
	"strconv"
	"sync"
	"time"

	"github.com/bluesky-social/indigo/atproto/atdata"

	"github.com/johnandersen777/socialweb-computer-kcp/internal/didkey"
	"github.com/johnandersen777/socialweb-computer-kcp/internal/kit/httpjson"
)

type Error struct {
	Status  int
	Message string
}

func (e *Error) Error() string { return e.Message }

func errorf(status int, format string, args ...any) *Error {
	return &Error{Status: status, Message: fmt.Sprintf(format, args...)}
}

type Options struct {
	Store           Store
	Version         string
	VerifySignature func(ctx context.Context, did string, data, sig []byte) (bool, error)
}

type Server struct {
	store     Store
	version   string
	verifySig func(ctx context.Context, did string, data, sig []byte) (bool, error)
	mu        sync.Mutex
	handler   http.Handler
}

func NewServer(opts Options) *Server {
	store := opts.Store
	if store == nil {
		store = NewMemoryStore()
	}
	version := opts.Version
	version = cmp.Or(version, "0.1.0")
	s := &Server{store: store, version: version, verifySig: opts.VerifySignature}
	if s.verifySig == nil {
		s.verifySig = func(ctx context.Context, did string, data, sig []byte) (bool, error) {
			err := didkey.Verify(did, data, sig)
			if err != nil {
				return false, err
			}
			return true, nil
		}
	}
	mux := http.NewServeMux()
	mux.HandleFunc("GET /health", s.handleHealth)
	mux.HandleFunc("GET /export", s.handleExport)
	mux.HandleFunc("GET /{did}", s.handleResolve)
	mux.HandleFunc("POST /{did}", s.handleSubmit)
	mux.HandleFunc("GET /{did}/log", s.handleLog)
	mux.HandleFunc("GET /{did}/log/audit", s.handleAudit)
	s.handler = withCORS(mux)
	return s
}

func (s *Server) Store() Store { return s.store }

func (s *Server) Handler() http.Handler { return s.handler }

func (s *Server) ServeHTTP(w http.ResponseWriter, r *http.Request) {
	s.handler.ServeHTTP(w, r)
}

func (s *Server) Submit(ctx context.Context, did string, raw map[string]any) (*LogEntry, error) {
	if _, ok := raw["type"]; !ok {
		return nil, errorf(400, "Operation must be an object")
	}
	if msg := ValidateStructure(raw); msg != "" {
		return nil, errorf(400, "%s", msg)
	}
	op := NewOperationFromObject(raw)

	s.mu.Lock()
	defer s.mu.Unlock()

	audit, err := s.store.GetAuditLog(ctx, did)
	if err != nil {
		return nil, err
	}
	current, err := s.store.GetCurrentOps(ctx, did)
	if err != nil {
		return nil, err
	}

	switch op.Type {
	case TypeCreate:
		if len(audit) > 0 {
			return nil, errorf(400, "Genesis operation requires empty log")
		}
	case TypeOperation, TypeTombstone:
		if op.Prev == nil && len(audit) > 0 {
			return nil, errorf(400, "Genesis operation requires empty log")
		}
		if err := validatePrevChain(op.Prev, current); err != nil {
			return nil, err
		}
		if len(current) > 0 {
			head := current[len(current)-1].Operation
			if head.Type == TypeTombstone {
				return nil, errorf(410, "DID not available: %s", did)
			}
			if head.Type == TypeOperation {
				if !s.verifySignatures(ctx, op, head.RotationKeys) {
					return nil, errorf(400, "Invalid Signature")
				}
			}
		} else {
			if !s.verifySignatures(ctx, op, op.RotationKeys) {
				return nil, errorf(400, "Invalid Signature")
			}
			if op.Type == TypeOperation {
				derived, err := OperationDID(op)
				if err != nil {
					return nil, errorf(400, "Failed to compute operation CID")
				}
				if derived != did {
					return nil, errorf(400, "DID does not match operation")
				}
			}
		}
	default:
		return nil, errorf(400, "Unknown operation type: %s", op.Type)
	}

	cid, err := OperationCID(op)
	if err != nil {
		return nil, errorf(400, "Failed to compute operation CID")
	}

	if op.Type == TypeOperation && op.Prev != nil && len(current) > 0 {
		prevIdx := -1
		for i := range current {
			if current[i].CID == *op.Prev {
				prevIdx = i
				break
			}
		}
		if prevIdx >= 0 && prevIdx < len(current)-1 {
			toNullify := make([]string, 0, len(current)-prevIdx-1)
			for _, e := range current[prevIdx+1:] {
				toNullify = append(toNullify, e.CID)
			}
			if err := s.store.NullifyOps(ctx, did, toNullify); err != nil {
				return nil, err
			}
		}
	}

	entry := LogEntry{
		DID:       did,
		Operation: *op,
		CID:       cid,
		Nullified: false,
		CreatedAt: time.Now().UTC().Format("2006-01-02T15:04:05.000Z"),
	}
	if err := s.store.InsertOp(ctx, entry); err != nil {
		return nil, err
	}
	return &entry, nil
}

func (s *Server) Resolve(ctx context.Context, did string) (*DIDDocument, error) {
	ops, err := s.store.GetCurrentOps(ctx, did)
	if err != nil {
		return nil, err
	}
	if len(ops) == 0 {
		return nil, errorf(404, "DID not registered: %s", did)
	}
	if ops[len(ops)-1].Operation.Type == TypeTombstone {
		return nil, errorf(410, "DID not available: %s", did)
	}
	doc := ResolveDocument(did, ops)
	if doc == nil {
		return nil, errorf(410, "DID not available: %s", did)
	}
	return doc, nil
}

func (s *Server) Log(ctx context.Context, did string) ([]Operation, error) {
	ops, err := s.store.GetCurrentOps(ctx, did)
	if err != nil {
		return nil, err
	}
	if len(ops) == 0 {
		return nil, errorf(404, "DID not registered: %s", did)
	}
	out := make([]Operation, len(ops))
	for i, e := range ops {
		out[i] = e.Operation
	}
	return out, nil
}

func (s *Server) AuditLog(ctx context.Context, did string) ([]LogEntry, error) {
	ops, err := s.store.GetAuditLog(ctx, did)
	if err != nil {
		return nil, err
	}
	if len(ops) == 0 {
		return nil, errorf(404, "DID not registered: %s", did)
	}
	return ops, nil
}

func (s *Server) Export(ctx context.Context, after *time.Time, count int) ([]LogEntry, error) {
	return s.store.ExportLogs(ctx, after, count)
}

func (s *Server) verifySignatures(ctx context.Context, op *Operation, keys []string) bool {
	sig, err := decodeSignature(op.Sig)
	if err != nil {
		return false
	}
	unsigned, err := atdata.MarshalCBOR(op.UnsignedObject())
	if err != nil {
		return false
	}
	for _, key := range keys {
		ok, err := s.verifySig(ctx, key, unsigned, sig)
		if err == nil && ok {
			return true
		}
	}
	return false
}

func validatePrevChain(prev *string, current []LogEntry) *Error {
	if prev == nil {
		if len(current) > 0 {
			return errorf(400, "Genesis operation requires empty log")
		}
		return nil
	}
	for _, e := range current {
		if e.CID == *prev {
			return nil
		}
	}
	return errorf(400, "prev CID not found in current chain: %s", *prev)
}

func (s *Server) handleHealth(w http.ResponseWriter, r *http.Request) {
	httpjson.Write(w, http.StatusOK, map[string]string{"version": s.version})
}

func (s *Server) handleExport(w http.ResponseWriter, r *http.Request) {
	q := r.URL.Query()
	var after *time.Time
	if v := q.Get("after"); v != "" {
		t, err := time.Parse(time.RFC3339Nano, v)
		if err != nil {
			httpjson.Write(w, http.StatusBadRequest, map[string]string{"message": "Invalid Query Parameter: after"})
			return
		}
		after = &t
	}
	count := -1
	if v := q.Get("count"); v != "" {
		n, err := strconv.Atoi(v)
		if err != nil || n < 0 {
			httpjson.Write(w, http.StatusBadRequest, map[string]string{"message": "Invalid Query Parameter: count"})
			return
		}
		count = min(n, 1000)
	}
	entries, err := s.store.ExportLogs(r.Context(), after, count)
	if err != nil {
		writeError(w, err)
		return
	}
	httpjson.Write(w, http.StatusOK, entries)
}

func (s *Server) handleResolve(w http.ResponseWriter, r *http.Request) {
	doc, err := s.Resolve(r.Context(), r.PathValue("did"))
	if err != nil {
		writeError(w, err)
		return
	}
	httpjson.Write(w, http.StatusOK, doc)
}

func (s *Server) handleSubmit(w http.ResponseWriter, r *http.Request) {
	body, err := io.ReadAll(io.LimitReader(r.Body, 4<<20))
	if err != nil {
		httpjson.Write(w, http.StatusBadRequest, map[string]string{"message": "Invalid JSON body"})
		return
	}
	var decoded any
	if err := json.Unmarshal(body, &decoded); err != nil {
		httpjson.Write(w, http.StatusBadRequest, map[string]string{"message": "Invalid JSON body"})
		return
	}
	raw, ok := decoded.(map[string]any)
	if !ok {
		httpjson.Write(w, http.StatusBadRequest, map[string]string{"message": "Operation must be an object"})
		return
	}
	entry, err := s.Submit(r.Context(), r.PathValue("did"), raw)
	if err != nil {
		writeError(w, err)
		return
	}
	httpjson.Write(w, http.StatusOK, entry)
}

func (s *Server) handleLog(w http.ResponseWriter, r *http.Request) {
	ops, err := s.Log(r.Context(), r.PathValue("did"))
	if err != nil {
		writeError(w, err)
		return
	}
	httpjson.Write(w, http.StatusOK, ops)
}

func (s *Server) handleAudit(w http.ResponseWriter, r *http.Request) {
	entries, err := s.AuditLog(r.Context(), r.PathValue("did"))
	if err != nil {
		writeError(w, err)
		return
	}
	httpjson.Write(w, http.StatusOK, entries)
}

func writeError(w http.ResponseWriter, err error) {
	var pe *Error
	if errors.As(err, &pe) {
		httpjson.Write(w, pe.Status, map[string]string{"message": pe.Message})
		return
	}
	httpjson.Write(w, http.StatusInternalServerError, map[string]string{"message": "Internal Server Error"})
}

func withCORS(next http.Handler) http.Handler {
	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		w.Header().Set("Access-Control-Allow-Origin", "*")
		w.Header().Set("Access-Control-Allow-Methods", "GET,HEAD,PUT,PATCH,POST,DELETE,OPTIONS")
		w.Header().Set("Access-Control-Allow-Headers", r.Header.Get("Access-Control-Request-Headers"))
		if r.Method == http.MethodOptions {
			w.WriteHeader(http.StatusNoContent)
			return
		}
		next.ServeHTTP(w, r)
	})
}