Merge pull request #21 from tablelandnetwork/bcalza/cache

adds cache support

cmd/basin/publication.go

@@ -59,7 +59,7 @@ func newPublicationCommand() *cli.Command {
 func newPublicationCreateCommand() *cli.Command {
 	var owner, dburi, provider string
 	var secure bool
-	var winSize int64
+	var winSize, cache int64
 
 	return &cli.Command{
 		Name:  "create",
@@ -94,6 +94,12 @@ func newPublicationCreateCommand() *cli.Command {
 				Destination: &winSize,
 				Value:       DefaultWindowSize,
 			},
+			&cli.Int64Flag{
+				Name:        "cache",
+				Usage:       "Time duration (in minutes) that the data will be available in the cache",
+				Destination: &cache,
+				Value:       0,
+			},
 		},
 		Action: func(cCtx *cli.Context) error {
 			if cCtx.NArg() != 1 {
@@ -147,7 +153,7 @@ func newPublicationCreateCommand() *cli.Command {
 				return fmt.Errorf("encode: %s", err)
 			}
 
-			exists, err := createPublication(cCtx.Context, dburi, ns, rel, provider, owner, secure)
+			exists, err := createPublication(cCtx.Context, dburi, ns, rel, provider, owner, secure, cache)
 			if err != nil {
 				return fmt.Errorf("failed to create publication: %s", err)
 			}
@@ -541,7 +547,7 @@ func newPublicationDealsCommand() *cli.Command {
 
 			if format == "table" {
 				table := tablewriter.NewWriter(os.Stdout)
-				table.SetHeader([]string{"CID", "Size", "Timestamp", "Archived"})
+				table.SetHeader([]string{"CID", "Size", "Timestamp", "Archived", "Cache Expiry"})
 
 				for _, deal := range deals {
 					isArchived := "N"
@@ -553,7 +559,7 @@ func newPublicationDealsCommand() *cli.Command {
 						timestamp = time.Unix(deal.Timestamp, 0).Format(time.RFC3339)
 					}
 					table.Append([]string{
-						deal.CID, fmt.Sprintf("%d", deal.Size), timestamp, isArchived,
+						deal.CID, fmt.Sprintf("%d", deal.Size), timestamp, isArchived, deal.CacheExpiry,
 					})
 				}
 				table.Render()
@@ -677,7 +683,14 @@ func inspectTable(ctx context.Context, tx pgx.Tx, rel string) ([]app.Column, err
 }
 
 func createPublication(
-	ctx context.Context, dburi string, ns string, rel string, provider string, owner string, secure bool,
+	ctx context.Context,
+	dburi string,
+	ns string,
+	rel string,
+	provider string,
+	owner string,
+	secure bool,
+	cacheDuration int64,
 ) (exists bool, err error) {
 	bp, err := basinprovider.New(ctx, provider, secure)
 	if err != nil {
@@ -686,7 +699,7 @@ func createPublication(
 	defer bp.Close()
 
 	if dburi == "" {
-		exists, err := bp.Create(ctx, ns, rel, basincapnp.Schema{}, common.HexToAddress(owner))
+		exists, err := bp.Create(ctx, ns, rel, basincapnp.Schema{}, common.HexToAddress(owner), cacheDuration)
 		if err != nil {
 			return false, fmt.Errorf("create call: %s", err)
 		}
@@ -751,7 +764,7 @@ func createPublication(
 		return false, fmt.Errorf("failed to create publication: %s", err)
 	}
 
-	if _, err := bp.Create(ctx, ns, rel, capnpSchema, common.HexToAddress(owner)); err != nil {
+	if _, err := bp.Create(ctx, ns, rel, capnpSchema, common.HexToAddress(owner), cacheDuration); err != nil {
 		return false, fmt.Errorf("create call: %s", err)
 	}
 

internal/app/basin_provider.go

@@ -10,15 +10,16 @@ import (
 
 // DealInfo represents information about a deal.
 type DealInfo struct {
-	CID        string `json:"cid"`
-	Timestamp  int64  `json:"timestamp"`
-	Size       uint32 `json:"size"`
-	IsArchived bool   `json:"is_archived"`
+	CID         string `json:"cid"`
+	Timestamp   int64  `json:"timestamp"`
+	Size        uint32 `json:"size"`
+	IsArchived  bool   `json:"is_archived"`
+	CacheExpiry string `json:"cache_expiry"`
 }
 
 // BasinProvider ...
 type BasinProvider interface {
-	Create(context.Context, string, string, basincapnp.Schema, common.Address) (bool, error)
+	Create(context.Context, string, string, basincapnp.Schema, common.Address, int64) (bool, error)
 	List(context.Context, common.Address) ([]string, error)
 	Deals(context.Context, string, string, uint32, uint64, Timestamp, Timestamp) ([]DealInfo, error)
 	LatestDeals(context.Context, string, string, uint32, Timestamp, Timestamp) ([]DealInfo, error)

internal/app/streamer_test.go

@@ -242,7 +242,7 @@ type basinProviderMock struct {
 }
 
 func (bp *basinProviderMock) Create(
-	_ context.Context, ns string, _ string, _ basincapnp.Schema, owner common.Address,
+	_ context.Context, ns string, _ string, _ basincapnp.Schema, owner common.Address, _ int64,
 ) (bool, error) {
 	bp.owner[ns] = owner.Hex()
 	return false, nil

pkg/basinprovider/provider.capnp

@@ -6,7 +6,7 @@ $Go.package("basinprovider");
 $Go.import("github.com/tablelandnetwork/basin-cli/pkg/basinprovider");
 
 interface Publications {
-	create @0 (ns :Text, rel :Text, schema :import "../capnp/definitions.capnp" .Schema, owner :Data) -> (exists :Bool);
+	create @0 (ns :Text, rel :Text, schema :import "../capnp/definitions.capnp" .Schema, owner :Data, cache_duration :Int64) -> (exists :Bool);
 	push @1 (ns :Text, rel :Text, tx :import "../capnp/definitions.capnp" .Tx, sig :Data);
 
 	upload @2 (ns :Text, rel :Text, size: UInt64, timestamp: Int64) -> (callback :Callback);