DDO with notifications spike

cmd/boostd/import_direct_data.go

@@ -1,15 +1,18 @@
 package main
 
 import (
+	"encoding/hex"
 	"fmt"
 	"os"
 	"path/filepath"
+	"strings"
 
 	bcli "github.com/filecoin-project/boost/cli"
 	"github.com/filecoin-project/boost/storagemarket/types"
 	"github.com/filecoin-project/go-address"
 	"github.com/filecoin-project/go-state-types/abi"
 	"github.com/filecoin-project/go-state-types/builtin"
+	"github.com/filecoin-project/go-state-types/builtin/v13/miner"
 	"github.com/filecoin-project/go-state-types/builtin/v9/verifreg"
 	lcli "github.com/filecoin-project/lotus/cli"
 	"github.com/google/uuid"
@@ -52,6 +55,10 @@ var importDirectDataCmd = &cli.Command{
 			Name:  "start-epoch",
 			Usage: "start epoch by when the deal should be proved by provider on-chain (default: 2 days from now)",
 		},
+		&cli.StringSliceFlag{
+			Name:  "notify",
+			Usage: "DDO notifications in format 'address:payload",
+		},
 	},
 	Action: func(cctx *cli.Context) error {
 		if cctx.Args().Len() < 2 {
@@ -127,6 +134,31 @@ var importDirectDataCmd = &cli.Command{
 		// Since StartEpoch is more than Head+StartEpochSealingBuffer, we can set end epoch as start+TermMin
 		endEpoch := startEpoch + alloc.TermMin
 
+		var notifications []miner.DataActivationNotification
+		for _, notifyStr := range cctx.StringSlice("notify") {
+			parts := strings.SplitN(notifyStr, ":", 2)
+			if len(parts) != 2 {
+				return fmt.Errorf("invalid notify format '%s': expected 'address:payload'", notifyStr)
+			}
+
+			// Parse address
+			addr, err := address.NewFromString(parts[0])
+			if err != nil {
+				return fmt.Errorf("invalid address in notify '%s': %w", parts[0], err)
+			}
+
+			// Parse payload (hex string)
+			payload, err := hex.DecodeString(strings.TrimPrefix(parts[1], "0x"))
+			if err != nil {
+				return fmt.Errorf("invalid hex payload in notify '%s': %w", parts[1], err)
+			}
+
+			notifications = append(notifications, miner.DataActivationNotification{
+				Address: addr,
+				Payload: payload,
+			})
+		}
+
 		ddParams := types.DirectDealParams{
 			DealUUID:           uuid.New(),
 			AllocationID:       verifreg.AllocationId(allocationId),
@@ -138,6 +170,7 @@ var importDirectDataCmd = &cli.Command{
 			DeleteAfterImport:  cctx.Bool("delete-after-import"),
 			RemoveUnsealedCopy: cctx.Bool("remove-unsealed-copy"),
 			SkipIPNIAnnounce:   cctx.Bool("skip-ipni-announce"),
+			Notifications:      notifications,
 		}
 		rej, err := napi.BoostDirectDeal(cctx.Context, ddParams)
 		if err != nil {

db/directdeals.go

@@ -66,6 +66,7 @@ func newDirectDealsAccessor(db *sql.DB, deal *types.DirectDeal) *directDealsAcce
 			"Retry":            &fielddef.FieldDef{F: &deal.Retry},
 			"KeepUnsealedCopy": &fielddef.FieldDef{F: &deal.KeepUnsealedCopy},
 			"AnnounceToIPNI":   &fielddef.FieldDef{F: &deal.AnnounceToIPNI},
+			"Notifications":    &fielddef.DDONotificationsFieldDef{F: deal.Notifications},
 		},
 	}
 }

db/fielddef/fielddef.go

@@ -1,6 +1,7 @@
 package fielddef
 
 import (
+	"bytes"
 	"database/sql"
 	"encoding/hex"
 	"fmt"
@@ -9,6 +10,7 @@ import (
 	"github.com/filecoin-project/go-address"
 	cborutil "github.com/filecoin-project/go-cbor-util"
 	"github.com/filecoin-project/go-state-types/big"
+	"github.com/filecoin-project/go-state-types/builtin/v13/miner"
 	"github.com/filecoin-project/go-state-types/builtin/v9/market"
 	"github.com/filecoin-project/go-state-types/crypto"
 	"github.com/ipfs/go-cid"
@@ -356,3 +358,34 @@ func (fd *SignedPropFieldDef) Unmarshall() error {
 	*fd.F = c
 	return nil
 }
+
+type DDONotificationsFieldDef struct {
+	Marshalled string
+	F          []miner.DataActivationNotification
+}
+
+func (fd *DDONotificationsFieldDef) FieldPtr() interface{} {
+	return &fd.Marshalled
+}
+
+func (fd *DDONotificationsFieldDef) Marshall() (interface{}, error) {
+	b, err := cborutil.Dump(fd.F)
+	if err != nil {
+		return nil, fmt.Errorf("failed to cbor dump DataActivationNotification: %w", err)
+	}
+	hexStr := fmt.Sprintf("%x", b)
+	return hexStr, nil
+}
+
+func (fd *DDONotificationsFieldDef) Unmarshall() error {
+	b, err := hex.DecodeString(fd.Marshalled)
+	if err != nil {
+		return fmt.Errorf("failed to decode hex string for DataActivationNotification: %w", err)
+	}
+	var out []miner.DataActivationNotification
+	if err := cborutil.ReadCborRPC(bytes.NewReader(b), &out); err != nil {
+		return fmt.Errorf("failed to cbor unmarshal DataActivationNotification: %w", err)
+	}
+	fd.F = out
+	return nil
+}

db/migrations/20230816133342_direct_deals_db.sql

@@ -22,6 +22,7 @@ CREATE TABLE IF NOT EXISTS DirectDeals (
     Retry TEXT,
     AnnounceToIPNI BOOL,
     KeepUnsealedCopy BOOL,
+    Notifications TEXT,
     PRIMARY KEY(ID)
 );
 -- +goose StatementEnd

storagemarket/direct_deals_provider.go

@@ -216,6 +216,7 @@ func (ddp *DirectDealsProvider) Import(ctx context.Context, params types.DirectD
 		AllocationID:     params.AllocationID,
 		KeepUnsealedCopy: !params.RemoveUnsealedCopy,
 		AnnounceToIPNI:   !params.SkipIPNIAnnounce,
+		Notifications:    params.Notifications,
 		Retry:            types.DealRetryAuto,
 	}
 
@@ -424,7 +425,7 @@ func (ddp *DirectDealsProvider) execDeal(ctx context.Context, entry *types.Direc
 					Client: abi.ActorID(clientId),
 					ID:     verifreg13types.AllocationId(entry.AllocationID),
 				},
-				Notify: nil,
+				Notify: entry.Notifications,
 			},
 
 			// Best-effort deal asks

storagemarket/types/direct_deal.go

@@ -7,6 +7,7 @@ import (
 	"github.com/filecoin-project/boost/storagemarket/types/dealcheckpoints"
 	"github.com/filecoin-project/go-address"
 	"github.com/filecoin-project/go-state-types/abi"
+	"github.com/filecoin-project/go-state-types/builtin/v13/miner"
 	verifregtypes "github.com/filecoin-project/go-state-types/builtin/v9/verifreg"
 	"github.com/google/uuid"
 	"github.com/ipfs/go-cid"
@@ -57,6 +58,9 @@ type DirectDeal struct {
 
 	//Announce deal to the IPNI(Index Provider)
 	AnnounceToIPNI bool
+
+	// Notify actors with payload info upon prove commit
+	Notifications []miner.DataActivationNotification
 }
 
 func (d *DirectDeal) String() string {

storagemarket/types/types.go

@@ -12,6 +12,7 @@ import (
 	"github.com/filecoin-project/boost/transport/types"
 	"github.com/filecoin-project/go-address"
 	"github.com/filecoin-project/go-state-types/abi"
+	"github.com/filecoin-project/go-state-types/builtin/v13/miner"
 	"github.com/filecoin-project/go-state-types/builtin/v9/market"
 	"github.com/filecoin-project/go-state-types/builtin/v9/verifreg"
 	"github.com/filecoin-project/go-state-types/crypto"
@@ -98,6 +99,7 @@ type DirectDealParams struct {
 	DeleteAfterImport  bool
 	RemoveUnsealedCopy bool
 	SkipIPNIAnnounce   bool
+	Notifications      []miner.DataActivationNotification
 }
 
 // Transfer has the parameters for a data transfer