stores: migrate ObjectsBySlabKey

stores/metadata.go

@@ -234,16 +234,6 @@ type (
 		FCID    fileContractID
 		HostKey publicKey
 	}
-
-	// rawObjectMetadata is used for hydrating object metadata.
-	rawObjectMetadata struct {
-		ETag       string
-		Health     float64
-		MimeType   string
-		ModTime    datetime
-		ObjectName string
-		Size       int64
-	}
 )
 
 func (s *contractState) LoadString(state string) error {
@@ -354,17 +344,6 @@ func (c dbContract) convert() api.ContractMetadata {
 	}
 }
 
-func (raw rawObjectMetadata) convert() api.ObjectMetadata {
-	return newObjectMetadata(
-		raw.ObjectName,
-		raw.ETag,
-		raw.MimeType,
-		raw.Health,
-		time.Time(raw.ModTime),
-		raw.Size,
-	)
-}
-
 func (raw rawObject) toSlabSlice() (slice object.SlabSlice, _ error) {
 	if len(raw) == 0 {
 		return object.SlabSlice{}, errors.New("no sectors found")
@@ -1245,32 +1224,10 @@ func (s *SQLStore) PackedSlabsForUpload(ctx context.Context, lockingDuration tim
 }
 
 func (s *SQLStore) ObjectsBySlabKey(ctx context.Context, bucket string, slabKey object.EncryptionKey) (metadata []api.ObjectMetadata, err error) {
-	var rows []rawObjectMetadata
-	key, err := slabKey.MarshalBinary()
-	if err != nil {
-		return nil, err
-	}
-
-	err = s.retryTransaction(ctx, func(tx *gorm.DB) error {
-		return tx.Raw(`
-SELECT DISTINCT obj.object_id as ObjectName, obj.size as Size, obj.mime_type as MimeType, sla.health as Health
-FROM slabs sla
-INNER JOIN slices sli ON sli.db_slab_id = sla.id
-INNER JOIN objects obj ON sli.db_object_id = obj.id
-INNER JOIN buckets b ON obj.db_bucket_id = b.id AND b.name = ?
-WHERE sla.key = ?
-	`, bucket, key).
-			Scan(&rows).
-			Error
+	err = s.db.Transaction(ctx, func(tx sql.DatabaseTx) error {
+		metadata, err = tx.ObjectsBySlabKey(ctx, bucket, slabKey)
+		return err
 	})
-	if err != nil {
-		return nil, err
-	}
-
-	// convert rows
-	for _, row := range rows {
-		metadata = append(metadata, row.convert())
-	}
 	return
 }
 

stores/sql/database.go

@@ -203,6 +203,10 @@ type (
 		// ObjectMetadata returns an object's metadata.
 		ObjectMetadata(ctx context.Context, bucket, key string) (api.Object, error)
 
+		// ObjectsBySlabKey returns all objects that contain a reference to the
+		// slab with the given slabKey.
+		ObjectsBySlabKey(ctx context.Context, bucket string, slabKey object.EncryptionKey) (metadata []api.ObjectMetadata, err error)
+
 		// ObjectsStats returns overall stats about stored objects
 		ObjectsStats(ctx context.Context, opts api.ObjectsStatsOpts) (api.ObjectsStatsResponse, error)
 

stores/sql/main.go

@@ -2657,3 +2657,36 @@ func SearchObjects(ctx context.Context, tx Tx, bucket, substring string, offset,
 	}
 	return objects, nil
 }
+
+func ObjectsBySlabKey(ctx context.Context, tx Tx, bucket string, slabKey object.EncryptionKey) ([]api.ObjectMetadata, error) {
+	key, err := slabKey.MarshalBinary()
+	if err != nil {
+		return nil, fmt.Errorf("failed to marshal encryption key: %w", err)
+	}
+	rows, err := tx.Query(ctx, fmt.Sprintf(`
+		SELECT %s
+		FROM objects o
+		INNER JOIN buckets b ON o.db_bucket_id = b.id
+		WHERE b.name = ? AND EXISTS (
+			SELECT 1
+			FROM objects o2
+			INNER JOIN slices sli ON sli.db_object_id = o2.id
+			INNER JOIN slabs sla ON sla.id = sli.db_slab_id
+			WHERE o2.id = o.id AND sla.key = ?
+		)
+	`, tx.SelectObjectMetadataExpr()), bucket, SecretKey(key))
+	if err != nil {
+		return nil, fmt.Errorf("failed to query objects: %w", err)
+	}
+	defer rows.Close()
+
+	var objects []api.ObjectMetadata
+	for rows.Next() {
+		om, err := tx.ScanObjectMetadata(rows)
+		if err != nil {
+			return nil, fmt.Errorf("failed to scan object metadata: %w", err)
+		}
+		objects = append(objects, om)
+	}
+	return objects, nil
+}

stores/sql/mysql/main.go

@@ -534,6 +534,10 @@ func (tx *MainDatabaseTx) ObjectMetadata(ctx context.Context, bucket, path strin
 	return ssql.ObjectMetadata(ctx, tx, bucket, path)
 }
 
+func (tx *MainDatabaseTx) ObjectsBySlabKey(ctx context.Context, bucket string, slabKey object.EncryptionKey) (metadata []api.ObjectMetadata, err error) {
+	return ssql.ObjectsBySlabKey(ctx, tx, bucket, slabKey)
+}
+
 func (tx *MainDatabaseTx) ObjectsStats(ctx context.Context, opts api.ObjectsStatsOpts) (api.ObjectsStatsResponse, error) {
 	return ssql.ObjectsStats(ctx, tx, opts)
 }

stores/sql/sqlite/main.go

@@ -531,6 +531,10 @@ func (tx *MainDatabaseTx) ObjectMetadata(ctx context.Context, bucket, path strin
 	return ssql.ObjectMetadata(ctx, tx, bucket, path)
 }
 
+func (tx *MainDatabaseTx) ObjectsBySlabKey(ctx context.Context, bucket string, slabKey object.EncryptionKey) (metadata []api.ObjectMetadata, err error) {
+	return ssql.ObjectsBySlabKey(ctx, tx, bucket, slabKey)
+}
+
 func (tx *MainDatabaseTx) ObjectsStats(ctx context.Context, opts api.ObjectsStatsOpts) (api.ObjectsStatsResponse, error) {
 	return ssql.ObjectsStats(ctx, tx, opts)
 }

stores/types.go

@@ -24,7 +24,6 @@ var zeroCurrency = currency(types.ZeroCurrency)
 
 type (
 	unixTimeMS     time.Time
-	datetime       time.Time
 	currency       types.Currency
 	bCurrency      types.Currency
 	fileContractID types.FileContractID
@@ -222,49 +221,6 @@ var SQLiteTimestampFormats = []string{
 	"2006-01-02",
 }
 
-// GormDataType implements gorm.GormDataTypeInterface.
-func (datetime) GormDataType() string {
-	return "string"
-}
-
-// Scan scan value into datetime, implements sql.Scanner interface.
-func (dt *datetime) Scan(value interface{}) error {
-	var s string
-	switch value := value.(type) {
-	case string:
-		s = value
-	case []byte:
-		s = string(value)
-	case time.Time:
-		*dt = datetime(value)
-		return nil
-	default:
-		return fmt.Errorf("failed to unmarshal time.Time value: %v %T", value, value)
-	}
-
-	var ok bool
-	var t time.Time
-	s = strings.TrimSuffix(s, "Z")
-	for _, format := range SQLiteTimestampFormats {
-		if timeVal, err := time.ParseInLocation(format, s, time.UTC); err == nil {
-			ok = true
-			t = timeVal
-			break
-		}
-	}
-	if !ok {
-		return fmt.Errorf("failed to parse datetime value: %v", s)
-	}
-
-	*dt = datetime(t)
-	return nil
-}
-
-// Value returns a datetime value, implements driver.Valuer interface.
-func (dt datetime) Value() (driver.Value, error) {
-	return (time.Time)(dt).Format(SQLiteTimestampFormats[0]), nil
-}
-
 // GormDataType implements gorm.GormDataTypeInterface.
 func (unixTimeMS) GormDataType() string {
 	return "BIGINT"