Add contract additions to worker cache (#1467)

It seems like right now the cached contracts are not updated when a new
contract is added. This fixes that.

I noticed because the worker failed to refill ephemeral accounts in my
open PR since the cached contracts were always 0 because the cache was
initialized before the contracts had been formed.

api/events.go

@@ -17,6 +17,7 @@ const (
 	ModuleHost        = "host"
 	ModuleSetting     = "setting"
 
+	EventAdd     = "add"
 	EventUpdate  = "update"
 	EventDelete  = "delete"
 	EventArchive = "archive"
@@ -34,6 +35,11 @@ type (
 		Timestamp      time.Time      `json:"timestamp"`
 	}
 
+	EventContractAdd struct {
+		Added     ContractMetadata `json:"added"`
+		Timestamp time.Time        `json:"timestamp"`
+	}
+
 	EventContractArchive struct {
 		ContractID types.FileContractID `json:"contractID"`
 		Reason     string               `json:"reason"`
@@ -79,6 +85,15 @@ var (
 		}
 	}
 
+	WebhookContractAdd = func(url string, headers map[string]string) webhooks.Webhook {
+		return webhooks.Webhook{
+			Event:   EventAdd,
+			Headers: headers,
+			Module:  ModuleContract,
+			URL:     url,
+		}
+	}
+
 	WebhookContractArchive = func(url string, headers map[string]string) webhooks.Webhook {
 		return webhooks.Webhook{
 			Event:   EventArchive,
@@ -142,6 +157,12 @@ func ParseEventWebhook(event webhooks.Event) (interface{}, error) {
 	switch event.Module {
 	case ModuleContract:
 		switch event.Event {
+		case EventAdd:
+			var e EventContractAdd
+			if err := json.Unmarshal(bytes, &e); err != nil {
+				return nil, err
+			}
+			return e, nil
 		case EventArchive:
 			var e EventContractArchive
 			if err := json.Unmarshal(bytes, &e); err != nil {

bus/routes.go

@@ -972,20 +972,29 @@ func (b *Bus) contractIDHandlerPOST(jc jape.Context) {
 	var req api.ContractAddRequest
 	if jc.DecodeParam("id", &id) != nil || jc.Decode(&req) != nil {
 		return
-	}
-	if req.Contract.ID() != id {
+	} else if req.Contract.ID() != id {
 		http.Error(jc.ResponseWriter, "contract ID mismatch", http.StatusBadRequest)
 		return
-	}
-	if req.TotalCost.IsZero() {
+	} else if req.TotalCost.IsZero() {
 		http.Error(jc.ResponseWriter, "TotalCost can not be zero", http.StatusBadRequest)
 		return
 	}
 
 	a, err := b.ms.AddContract(jc.Request.Context(), req.Contract, req.ContractPrice, req.TotalCost, req.StartHeight, req.State)
-	if jc.Check("couldn't store contract", err) == nil {
-		jc.Encode(a)
+	if jc.Check("couldn't store contract", err) != nil {
+		return
 	}
+
+	b.broadcastAction(webhooks.Event{
+		Module: api.ModuleContract,
+		Event:  api.EventAdd,
+		Payload: api.EventContractAdd{
+			Added:     a,
+			Timestamp: time.Now().UTC(),
+		},
+	})
+
+	jc.Encode(a)
 }
 
 func (b *Bus) contractIDRenewedHandlerPOST(jc jape.Context) {

internal/worker/cache.go

@@ -168,6 +168,9 @@ func (c *cache) HandleEvent(event webhooks.Event) (err error) {
 	case api.EventConsensusUpdate:
 		log = log.With("bh", e.BlockHeight, "ts", e.Timestamp)
 		c.handleConsensusUpdate(e)
+	case api.EventContractAdd:
+		log = log.With("fcid", e.Added.ID, "ts", e.Timestamp)
+		c.handleContractAdd(e)
 	case api.EventContractArchive:
 		log = log.With("fcid", e.ContractID, "ts", e.Timestamp)
 		c.handleContractArchive(e)
@@ -234,6 +237,24 @@ func (c *cache) handleConsensusUpdate(event api.EventConsensusUpdate) {
 	c.cache.Set(cacheKeyGougingParams, gp)
 }
 
+func (c *cache) handleContractAdd(event api.EventContractAdd) {
+	// return early if the cache doesn't have contracts
+	value, found, _ := c.cache.Get(cacheKeyDownloadContracts)
+	if !found {
+		return
+	}
+	contracts := value.([]api.ContractMetadata)
+
+	// add the contract to the cache
+	for _, contract := range contracts {
+		if contract.ID == event.Added.ID {
+			return
+		}
+	}
+	contracts = append(contracts, event.Added)
+	c.cache.Set(cacheKeyDownloadContracts, contracts)
+}
+
 func (c *cache) handleContractArchive(event api.EventContractArchive) {
 	// return early if the cache doesn't have contracts
 	value, found, _ := c.cache.Get(cacheKeyDownloadContracts)

internal/worker/events.go

@@ -112,6 +112,7 @@ func (e *eventSubscriber) Register(ctx context.Context, eventsURL string, opts .
 	// prepare webhooks
 	webhooks := []webhooks.Webhook{
 		api.WebhookConsensusUpdate(eventsURL, headers),
+		api.WebhookContractAdd(eventsURL, headers),
 		api.WebhookContractArchive(eventsURL, headers),
 		api.WebhookContractRenew(eventsURL, headers),
 		api.WebhookHostUpdate(eventsURL, headers),

internal/worker/events_test.go

@@ -156,8 +156,8 @@ func TestEventSubscriber(t *testing.T) {
 	time.Sleep(testRegisterInterval)
 
 	// assert webhook was registered
-	if webhooks := w.Webhooks(); len(webhooks) != 5 {
-		t.Fatal("expected 5 webhooks, got", len(webhooks))
+	if webhooks := w.Webhooks(); len(webhooks) != 6 {
+		t.Fatal("expected 6 webhooks, got", len(webhooks))
 	}
 
 	// send the same event again