Statuses

Author: aopoltorzhicky

go.mod

@@ -17,6 +17,7 @@ require (
 	github.com/goccy/go-json v0.10.5
 	github.com/opus-domini/fast-shot v1.1.4
 	github.com/pkg/errors v0.9.1
+	github.com/rs/zerolog v1.33.0
 	github.com/stretchr/testify v1.10.0
 	github.com/uptrace/bun v1.1.17
 	go.uber.org/mock v0.5.0
@@ -94,7 +95,6 @@ require (
 	github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect
 	github.com/power-devops/perfstat v0.0.0-20210106213030-5aafc221ea8c // indirect
 	github.com/rogpeppe/go-internal v1.12.0 // indirect
-	github.com/rs/zerolog v1.33.0 // indirect
 	github.com/shirou/gopsutil/v3 v3.23.12 // indirect
 	github.com/shoenig/go-m1cpu v0.1.6 // indirect
 	github.com/sirupsen/logrus v1.9.3 // indirect

pkg/api/types.go

@@ -10,4 +10,5 @@ type Change struct {
 	Address     string `json:"address"`
 	ImageURL    string `json:"image_url,omitempty"`
 	ChangeID    int64  `json:"change_id"`
+	Status      string `json:"status"`
 }

pkg/module/module.go

@@ -20,11 +20,12 @@ import (
 type Module struct {
 	modules.BaseModule
 
-	celestials celestials.API
-	address    IdByHash
-	states     storage.ICelestialState
-	tx         sdk.Transactable
-	state      storage.CelestialState
+	celestialsApi celestials.API
+	address       IdByHash
+	states        storage.ICelestialState
+	celestials    storage.ICelestial
+	tx            sdk.Transactable
+	state         storage.CelestialState
 
 	celestialsDatasource config.DataSource
 	indexerName          string
@@ -38,6 +39,7 @@ type Module struct {
 func New(
 	celestialsDatasource config.DataSource,
 	address IdByHash,
+	celestials storage.ICelestial,
 	state storage.ICelestialState,
 	tx sdk.Transactable,
 	indexerName string,
@@ -47,9 +49,10 @@ func New(
 	module := Module{
 		BaseModule:           modules.New("celestials"),
 		address:              address,
+		celestials:           celestials,
 		states:               state,
 		tx:                   tx,
-		celestials:           v1.New(celestialsDatasource.URL),
+		celestialsApi:        v1.New(celestialsDatasource.URL),
 		indexerName:          indexerName,
 		network:              network,
 		indexPeriod:          time.Minute,
@@ -124,7 +127,7 @@ func (m *Module) getChanges(ctx context.Context) (celestials.Changes, error) {
 	requestCtx, cancel := context.WithTimeout(ctx, time.Second*time.Duration(m.celestialsDatasource.Timeout))
 	defer cancel()
 
-	return m.celestials.Changes(
+	return m.celestialsApi.Changes(
 		requestCtx,
 		m.network,
 		celestials.WithFromChangeId(m.state.ChangeId),
@@ -145,6 +148,7 @@ func (m *Module) sync(ctx context.Context) error {
 		}
 
 		cids := make(map[string]storage.Celestial)
+		addressIds := make(map[uint64]struct{})
 
 		for i := range changes.Changes {
 			if m.state.ChangeId >= changes.Changes[i].ChangeID {
@@ -169,15 +173,25 @@ func (m *Module) sync(ctx context.Context) error {
 				return errors.Errorf("can't find address %s", changes.Changes[i].Address)
 			}
 
+			status, err := storage.ParseStatus(changes.Changes[i].Status)
+			if err != nil {
+				return err
+			}
+
+			if status == storage.StatusPRIMARY {
+				addressIds[addressId[0]] = struct{}{}
+			}
+
 			cids[changes.Changes[i].CelestialID] = storage.Celestial{
 				Id:        changes.Changes[i].CelestialID,
 				ImageUrl:  changes.Changes[i].ImageURL,
 				AddressId: addressId[0],
 				ChangeId:  changes.Changes[i].ChangeID,
+				Status:    status,
 			}
 		}
 
-		if err := m.save(ctx, cids); err != nil {
+		if err := m.save(ctx, cids, addressIds); err != nil {
 			return errors.Wrap(err, "save")
 		}
 		end = len(changes.Changes) < int(m.limit)
@@ -187,7 +201,7 @@ func (m *Module) sync(ctx context.Context) error {
 	return nil
 }
 
-func (m *Module) save(ctx context.Context, cids map[string]storage.Celestial) error {
+func (m *Module) save(ctx context.Context, cids map[string]storage.Celestial, addressIds map[uint64]struct{}) error {
 	requestCtx, cancel := context.WithTimeout(ctx, m.databaseTimeout)
 	defer cancel()
 
@@ -197,6 +211,10 @@ func (m *Module) save(ctx context.Context, cids map[string]storage.Celestial) er
 	}
 	defer tx.Close(requestCtx)
 
+	if err := tx.UpdateStatusForAddress(ctx, maps.Keys(addressIds)); err != nil {
+		return tx.HandleError(requestCtx, errors.Wrap(err, "update primary statuses"))
+	}
+
 	if err := tx.SaveCelestials(requestCtx, maps.Values(cids)); err != nil {
 		return tx.HandleError(requestCtx, errors.Wrap(err, "save celestials"))
 	}

pkg/module/module_test.go

@@ -51,6 +51,9 @@ func (s *ModuleTestSuite) SetupSuite() {
 	s.psqlContainer = psqlContainer
 
 	init := func(ctx context.Context, conn *database.Bun) error {
+		if err := pg.CreateTypes(ctx, conn); err != nil {
+			return err
+		}
 		if err := database.CreateTables(ctx, conn, new(storage.Celestial), new(storage.CelestialState)); err != nil {
 			if err := conn.Close(); err != nil {
 				return err
@@ -118,6 +121,7 @@ func (s *ModuleTestSuite) TestSync() {
 					Address:     "celestia1mm8yykm46ec3t0dgwls70g0jvtm055wk9ayal8",
 					ImageURL:    "image_url",
 					ChangeID:    4,
+					Status:      "PRIMARY",
 				},
 			},
 		}, nil)
@@ -140,14 +144,15 @@ func (s *ModuleTestSuite) TestSync() {
 	m := New(
 		cfgDs,
 		s.address,
+		s.celestials,
 		s.celestialState,
 		s.storage.Transactable,
 		testIndexerName,
 		network,
 		WithAddressPrefix("celestia"),
 		WithLimit(10),
 	)
-	m.celestials = s.api
+	m.celestialsApi = s.api
 
 	err = m.getState(ctx)
 	s.Require().NoError(err)

pkg/storage/celestial.go

@@ -11,15 +11,17 @@ import (
 type ICelestial interface {
 	ById(ctx context.Context, id string) (Celestial, error)
 	ByAddressId(ctx context.Context, addressId uint64, limit, offset int) ([]Celestial, error)
+	Primary(ctx context.Context, addressId uint64) (Celestial, error)
 }
 
 type Celestial struct {
 	bun.BaseModel `bun:"celestial" comment:"Table with celestial ids."`
 
-	Id        string `bun:"id,pk,notnull" comment:"Celestial id"`
-	AddressId uint64 `bun:"address_id"    comment:"Internal address identity for connected address"`
-	ImageUrl  string `bun:"image_url"     comment:"Image url"`
-	ChangeId  int64  `bun:"change_id"     comment:"Id of the last change of celestial id"`
+	Id        string `bun:"id,pk,notnull"                 comment:"Celestial id"`
+	AddressId uint64 `bun:"address_id"                    comment:"Internal address identity for connected address"`
+	ImageUrl  string `bun:"image_url"                     comment:"Image url"`
+	ChangeId  int64  `bun:"change_id"                     comment:"Id of the last change of celestial id"`
+	Status    Status `bun:"status,type:celestials_status" comment:"Status of celestial domain"`
 }
 
 func (Celestial) TableName() string {

pkg/storage/mock/celestial.go

@@ -40,3 +40,12 @@ func (c *Celestials) ByAddressId(ctx context.Context, addressId uint64, limit, o
 	err = query.Limit(limit).Scan(ctx)
 	return
 }
+
+func (c *Celestials) Primary(ctx context.Context, addressId uint64) (result storage.Celestial, err error) {
+	err = c.DB().NewSelect().
+		Model(&result).
+		Where("address_id = ?", addressId).
+		Where("status = ?", storage.StatusPRIMARY).
+		Scan(ctx)
+	return
+}