feat: server implementation

Author: alanshaw

client/retrieval/client.go

@@ -0,0 +1,102 @@
+package retrieval
+
+import (
+	"context"
+	"crypto/sha256"
+	"errors"
+	"fmt"
+	"hash"
+	"net/http"
+	"net/url"
+
+	"github.com/storacha/go-ucanto/client"
+	"github.com/storacha/go-ucanto/core/invocation"
+	"github.com/storacha/go-ucanto/core/message"
+	"github.com/storacha/go-ucanto/transport"
+	"github.com/storacha/go-ucanto/transport/headercar"
+	thttp "github.com/storacha/go-ucanto/transport/http"
+	"github.com/storacha/go-ucanto/ucan"
+)
+
+// NewConnection creates a new connection to a retrieval server that uses the
+// headercar transport.
+func NewConnection(id ucan.Principal, endpoint string) (*Connection, error) {
+	hasher := sha256.New
+	url, err := url.Parse(endpoint)
+	if err != nil {
+		return nil, err
+	}
+	channel := thttp.NewChannel(
+		url,
+		thttp.WithMethod("GET"),
+		thttp.WithSuccessStatusCode(
+			http.StatusOK,
+			http.StatusPartialContent,
+			http.StatusNoContent,
+			http.StatusNotExtended,
+		),
+	)
+	codec := headercar.NewOutboundCodec()
+	return &Connection{id, channel, codec, hasher}, nil
+}
+
+type Connection struct {
+	id      ucan.Principal
+	channel transport.Channel
+	codec   transport.OutboundCodec
+	hasher  func() hash.Hash
+}
+
+var _ client.Connection = (*Connection)(nil)
+
+func (c *Connection) ID() ucan.Principal {
+	return c.id
+}
+
+func (c *Connection) Codec() transport.OutboundCodec {
+	return c.codec
+}
+
+func (c *Connection) Channel() transport.Channel {
+	return c.channel
+}
+
+func (c *Connection) Hasher() hash.Hash {
+	return c.hasher()
+}
+
+func Execute(ctx context.Context, inv invocation.Invocation, conn client.Connection) (client.ExecutionResponse, transport.HTTPResponse, error) {
+	input, err := message.Build([]invocation.Invocation{inv}, nil)
+	if err != nil {
+		return nil, nil, fmt.Errorf("building message: %w", err)
+	}
+
+	var response transport.HTTPResponse
+	for {
+		req, err := conn.Codec().Encode(input)
+		if err != nil {
+			return nil, nil, fmt.Errorf("encoding message: %w", err)
+		}
+
+		// TODO: split request if too large
+
+		res, err := conn.Channel().Request(ctx, req)
+		if err != nil {
+			return nil, nil, fmt.Errorf("sending message: %w", err)
+		}
+
+		if res.Status() == http.StatusNotExtended {
+			return nil, nil, errors.New("not implemented")
+		}
+
+		response = res
+		break
+	}
+
+	output, err := conn.Codec().Decode(response)
+	if err != nil {
+		return nil, nil, fmt.Errorf("decoding message: %w", err)
+	}
+
+	return client.ExecutionResponse(output), response, nil
+}

retrieval/client/connection.go

@@ -0,0 +1,39 @@
+package datamodel
+
+import (
+	// for go:embed
+	_ "embed"
+	"fmt"
+
+	"github.com/ipld/go-ipld-prime"
+	"github.com/ipld/go-ipld-prime/schema"
+)
+
+//go:embed errors.ipldsch
+var errorsch []byte
+
+var (
+	errorTypeSystem *schema.TypeSystem
+)
+
+func init() {
+	ts, err := ipld.LoadSchemaBytes(errorsch)
+	if err != nil {
+		panic(fmt.Errorf("failed to load IPLD schema: %w", err))
+	}
+	errorTypeSystem = ts
+}
+
+func Schema() []byte {
+	return errorsch
+}
+
+func MissingProofsType() schema.Type {
+	return errorTypeSystem.TypeByName("MissingProofs")
+}
+
+type MissingProofsModel struct {
+	Name    string
+	Message string
+	Proofs  []ipld.Link
+}

server/retrieval/datamodel/errors.go

@@ -0,0 +1,5 @@
+type MissingProofs struct {
+	name String
+	message String
+	proofs [Link]
+}

server/retrieval/datamodel/errors.ipldsch

@@ -0,0 +1,61 @@
+package retrieval
+
+import (
+	"fmt"
+	"strings"
+
+	"github.com/storacha/go-ucanto/core/ipld"
+	"github.com/storacha/go-ucanto/core/result/failure"
+	rdm "github.com/storacha/go-ucanto/server/retrieval/datamodel"
+)
+
+type AgentMessageInvocationError struct{}
+
+func (amie AgentMessageInvocationError) Error() string {
+	return "Agent Message is required to have a single invocation."
+}
+
+func (amie AgentMessageInvocationError) Name() string {
+	return "AgentMessageInvocationError"
+}
+
+func (amie AgentMessageInvocationError) ToIPLD() (ipld.Node, error) {
+	return failure.FromError(amie).ToIPLD()
+}
+
+func NewAgentMessageInvocationError() AgentMessageInvocationError {
+	return AgentMessageInvocationError{}
+}
+
+type MissingProofs struct {
+	proofs []ipld.Link
+}
+
+func (mpe MissingProofs) Error() string {
+	var links []string
+	for _, p := range mpe.proofs {
+		links = append(links, p.String())
+	}
+	return fmt.Sprintf("Missing proofs: %s", strings.Join(links, ", "))
+}
+
+func (mpe MissingProofs) Name() string {
+	return "MissingProofs"
+}
+
+func (mpe MissingProofs) Proofs() []ipld.Link {
+	return mpe.proofs
+}
+
+func (mpe MissingProofs) ToIPLD() (ipld.Node, error) {
+	mdl := rdm.MissingProofsModel{
+		Name:    mpe.Name(),
+		Message: mpe.Error(),
+		Proofs:  mpe.Proofs(),
+	}
+	return ipld.WrapWithRecovery(&mdl, rdm.MissingProofsType())
+}
+
+func NewMissingProofsError(proofs []ipld.Link) MissingProofs {
+	return MissingProofs{proofs}
+}

server/retrieval/error.go

@@ -6,27 +6,38 @@ import (
 	"github.com/storacha/go-ucanto/core/invocation"
 	"github.com/storacha/go-ucanto/core/ipld"
 	"github.com/storacha/go-ucanto/core/receipt/fx"
+	"github.com/storacha/go-ucanto/core/result"
+	"github.com/storacha/go-ucanto/core/result/failure"
 	"github.com/storacha/go-ucanto/server"
 	"github.com/storacha/go-ucanto/server/transaction"
 	"github.com/storacha/go-ucanto/ucan"
 	"github.com/storacha/go-ucanto/validator"
 )
 
 // HandlerFunc is an invocation handler function. It is different to
-// [server.HandlerFunc] in that it allows an [RetrievalResponse] to be returned,
+// [server.HandlerFunc] in that it allows an [Response] to be returned,
 // which for a retrieval server will determine the HTTP headers and body content
 // of the HTTP response. The usual handler response (out and effects) are added
 // to the X-Agent-Message HTTP header.
-type HandlerFunc[C any, O ipld.Builder] func(ctx context.Context, capability ucan.Capability[C], invocation invocation.Invocation, context server.InvocationContext) (O, fx.Effects, *RetrievalResponse, error)
+type HandlerFunc[C any, O ipld.Builder, X failure.IPLDBuilderFailure] func(
+	ctx context.Context,
+	capability ucan.Capability[C],
+	invocation invocation.Invocation,
+	context server.InvocationContext,
+	request Request,
+) (result result.Result[O, X], fx fx.Effects, resp Response, err error)
 
 // Provide is used to define given capability provider. It decorates the passed
 // handler and takes care of UCAN validation. It only calls the handler
 // when validation succeeds.
-func Provide[C any, O ipld.Builder](capability validator.CapabilityParser[C], handler HandlerFunc[C, O]) ServiceMethod[O] {
-	return func(ctx context.Context, inv invocation.Invocation, ictx server.InvocationContext) (transaction.Transaction[O, ipld.Builder], *RetrievalResponse, error) {
-		var response *RetrievalResponse
-		method := server.Provide(capability, func(ctx context.Context, capability ucan.Capability[C], inv invocation.Invocation, ictx server.InvocationContext) (O, fx.Effects, error) {
-			out, fx, res, err := handler(ctx, capability, inv, ictx)
+func Provide[C any, O ipld.Builder, X failure.IPLDBuilderFailure](
+	capability validator.CapabilityParser[C],
+	handler HandlerFunc[C, O, X],
+) ServiceMethod[O, failure.IPLDBuilderFailure] {
+	return func(ctx context.Context, inv invocation.Invocation, ictx server.InvocationContext, req Request) (transaction.Transaction[O, failure.IPLDBuilderFailure], Response, error) {
+		var response Response
+		method := server.Provide(capability, func(ctx context.Context, capability ucan.Capability[C], inv invocation.Invocation, ictx server.InvocationContext) (result.Result[O, X], fx.Effects, error) {
+			out, fx, res, err := handler(ctx, capability, inv, ictx, req)
 			response = res
 			return out, fx, err
 		})

server/retrieval/handler.go

@@ -7,16 +7,15 @@ import (
 	"github.com/storacha/go-ucanto/core/invocation"
 	"github.com/storacha/go-ucanto/core/ipld"
 	"github.com/storacha/go-ucanto/core/result"
+	"github.com/storacha/go-ucanto/core/result/failure"
 	"github.com/storacha/go-ucanto/server"
 	"github.com/storacha/go-ucanto/server/transaction"
 	"github.com/storacha/go-ucanto/ucan"
 	"github.com/storacha/go-ucanto/validator"
 )
 
 // Option is an option configuring a ucanto retrieval server. It does not
-// include a transport codec option as it must be [headercar]. Service methods
-// also have a different signature to allow them to return response body data,
-// HTTP status codes and HTTP headers.
+// include a transport codec option as it must be [headercar].
 type Option func(cfg *srvConfig) error
 
 type srvConfig struct {
@@ -32,15 +31,19 @@ type srvConfig struct {
 	delegationCache       delegation.Store
 }
 
-func WithServiceMethod[O ipld.Builder](can string, handleFunc ServiceMethod[O]) Option {
+func WithServiceMethod[O ipld.Builder, X failure.IPLDBuilderFailure](can string, handler ServiceMethod[O, X]) Option {
 	return func(cfg *srvConfig) error {
-		cfg.service[can] = func(ctx context.Context, input invocation.Invocation, ictx server.InvocationContext, request *RetrievalRequest) (transaction.Transaction[ipld.Builder, ipld.Builder], *RetrievalResponse, error) {
-			tx, res, err := handleFunc(ctx, input, ictx, request)
+		cfg.service[can] = func(ctx context.Context, input invocation.Invocation, invCtx server.InvocationContext, req Request) (transaction.Transaction[ipld.Builder, failure.IPLDBuilderFailure], Response, error) {
+			tx, resp, err := handler(ctx, input, invCtx, req)
 			if err != nil {
-				return nil, nil, err
+				return nil, resp, err
 			}
-			out := result.MapOk(tx.Out(), func(o O) ipld.Builder { return o })
-			return transaction.NewTransaction(out, transaction.WithEffects(tx.Fx())), res, nil
+			out := result.MapResultR0(
+				tx.Out(),
+				func(o O) ipld.Builder { return o },
+				func(x X) failure.IPLDBuilderFailure { return x },
+			)
+			return transaction.NewTransaction(out, transaction.WithEffects(tx.Fx())), resp, nil
 		}
 		return nil
 	}