libs5-go/service/default/storage.go

326 lines
7.4 KiB
Go
Raw Normal View History

package _default
import (
2024-02-27 08:28:25 +00:00
"context"
"errors"
"fmt"
"git.lumeweb.com/LumeWeb/libs5-go/encoding"
"git.lumeweb.com/LumeWeb/libs5-go/metadata"
"git.lumeweb.com/LumeWeb/libs5-go/service"
"git.lumeweb.com/LumeWeb/libs5-go/storage"
"git.lumeweb.com/LumeWeb/libs5-go/storage/provider"
"git.lumeweb.com/LumeWeb/libs5-go/structs"
"git.lumeweb.com/LumeWeb/libs5-go/types"
"git.lumeweb.com/LumeWeb/libs5-go/utils"
"github.com/ddo/rq"
_ "github.com/ddo/rq"
"github.com/vmihailenco/msgpack/v5"
"go.etcd.io/bbolt"
"go.uber.org/zap"
"io"
"net/http"
"time"
)
const cacheBucketName = "object-cache"
var (
_ service.Service = (*StorageService)(nil)
_ service.StorageService = (*StorageService)(nil)
)
type StorageService struct {
metadataCache structs.Map
providerStore storage.ProviderStore
service.ServiceBase
}
func NewStorage(params service.ServiceParams) *StorageService {
return &StorageService{
metadataCache: structs.NewMap(),
ServiceBase: service.NewServiceBase(params.Logger, params.Config, params.Db),
}
}
2024-02-27 08:28:25 +00:00
func (s *StorageService) Start(ctx context.Context) error {
err :=
utils.CreateBucket(cacheBucketName, s.Db())
if err != nil {
return err
}
return nil
}
2024-02-27 08:28:25 +00:00
func (s *StorageService) Stop(ctx context.Context) error {
return nil
}
2024-02-27 08:28:25 +00:00
func (s *StorageService) Init(ctx context.Context) error {
return nil
}
func (n *StorageService) SetProviderStore(store storage.ProviderStore) {
n.providerStore = store
}
func (n *StorageService) ProviderStore() storage.ProviderStore {
return n.providerStore
}
func (s *StorageService) GetCachedStorageLocations(hash *encoding.Multihash, kinds []types.StorageLocationType) (map[string]storage.StorageLocation, error) {
locations := make(map[string]storage.StorageLocation)
locationMap, err := s.readStorageLocationsFromDB(hash)
if err != nil {
return nil, err
}
if len(locationMap) == 0 {
return make(map[string]storage.StorageLocation), nil
}
ts := time.Now().Unix()
for _, t := range kinds {
nodeMap, ok := (locationMap)[int(t)]
if !ok {
continue
}
for key, value := range nodeMap {
expiry, ok := value[3].(int64)
if !ok || expiry < ts {
continue
}
addressesInterface, ok := value[1].([]interface{})
if !ok {
continue
}
// Create a slice to hold the strings
addresses := make([]string, len(addressesInterface))
// Convert each element to string
for i, v := range addressesInterface {
str, ok := v.(string)
if !ok {
// Handle the error, maybe skip this element or set a default value
continue
}
addresses[i] = str
}
storageLocation := storage.NewStorageLocation(int(t), addresses, expiry)
if providerMessage, ok := value[4].([]byte); ok {
(storageLocation).SetProviderMessage(providerMessage)
}
locations[key] = storageLocation
}
}
return locations, nil
}
func (s *StorageService) readStorageLocationsFromDB(hash *encoding.Multihash) (storage.StorageLocationMap, error) {
var locationMap storage.StorageLocationMap
err := s.Db().View(func(tx *bbolt.Tx) error {
b := tx.Bucket([]byte(cacheBucketName)) // Replace with your actual bucket name
if b == nil {
return fmt.Errorf("bucket %s not found", cacheBucketName)
}
bytes := b.Get(hash.FullBytes())
if bytes == nil {
// If no data found, return an empty locationMap but no error
locationMap = storage.NewStorageLocationMap()
return nil
}
return msgpack.Unmarshal(bytes, &locationMap)
})
if err != nil {
return nil, err
}
return locationMap, nil
}
func (s *StorageService) AddStorageLocation(hash *encoding.Multihash, nodeId *encoding.NodeId, location storage.StorageLocation, message []byte) error {
// Read existing storage locations
locationDb, err := s.readStorageLocationsFromDB(hash)
if err != nil {
return err
}
nodeIdStr, err := nodeId.ToString()
if err != nil {
return err
}
// Get or create the inner map for the specific type
innerMap, exists := locationDb[location.Type()]
if !exists {
innerMap = make(storage.NodeStorage, 1)
innerMap[nodeIdStr] = make(storage.NodeDetailsStorage, 1)
}
// Create location map with new data
locationMap := make(map[int]interface{}, 3)
locationMap[1] = location.Parts()
locationMap[3] = location.Expiry()
locationMap[4] = message
// Update the inner map with the new location
innerMap[nodeIdStr] = locationMap
locationDb[location.Type()] = innerMap
// Serialize the updated map and store it in the database
packedBytes, err := msgpack.Marshal(locationDb)
if err != nil {
return err
}
err = s.Db().Update(func(tx *bbolt.Tx) error {
b := tx.Bucket([]byte(cacheBucketName))
return b.Put(hash.FullBytes(), packedBytes)
})
if err != nil {
return err
}
return nil
}
func (s *StorageService) DownloadBytesByHash(hash *encoding.Multihash) ([]byte, error) {
// Initialize the download URI provider
dlUriProvider := provider.NewStorageLocationProvider(provider.StorageLocationProviderParams{
Services: s.Services(),
Hash: hash,
LocationTypes: []types.StorageLocationType{
types.StorageLocationTypeFull,
types.StorageLocationTypeFile,
},
ServiceParams: service.ServiceParams{
Logger: s.Logger(),
Config: s.Config(),
Db: s.Db(),
},
})
err := dlUriProvider.Start()
if err != nil {
return nil, err
}
retryCount := 0
for {
dlUri, err := dlUriProvider.Next()
if err != nil {
return nil, err
}
s.Logger().Debug("Trying to download from", zap.String("url", dlUri.Location().BytesURL()))
req := rq.Get(dlUri.Location().BytesURL())
httpReq, err := req.ParseRequest()
if err != nil {
return nil, err
}
res, err := http.DefaultClient.Do(httpReq)
if err != nil {
err := dlUriProvider.Downvote(dlUri)
if err != nil {
return nil, err
}
retryCount++
if retryCount > 32 {
return nil, errors.New("too many retries")
}
continue
}
defer func(Body io.ReadCloser) {
err := Body.Close()
if err != nil {
s.Logger().Error("error closing body", zap.Error(err))
}
}(res.Body)
if res.StatusCode != 200 {
err := dlUriProvider.Downvote(dlUri)
if err != nil {
return nil, err
}
continue
}
bodyBytes, err := io.ReadAll(res.Body)
if err != nil {
return nil, err
}
return bodyBytes, nil
}
}
func (s *StorageService) DownloadBytesByCID(cid *encoding.CID) (bytes []byte, err error) {
bytes, err = s.DownloadBytesByHash(&cid.Hash)
if err != nil {
return nil, err
}
return bytes, nil
}
func (s *StorageService) GetMetadataByCID(cid *encoding.CID) (md metadata.Metadata, err error) {
hashStr, err := cid.Hash.ToString()
if err != nil {
return nil, err
}
if s.metadataCache.Contains(hashStr) {
md, _ := s.metadataCache.Get(hashStr)
return md.(metadata.Metadata), nil
}
bytes, err := s.DownloadBytesByHash(&cid.Hash)
if err != nil {
return nil, err
}
switch cid.Type {
case types.CIDTypeMetadataMedia, types.CIDTypeBridge: // Both cases use the same deserialization method
md = metadata.NewEmptyMediaMetadata()
err = msgpack.Unmarshal(bytes, md)
if err != nil {
return nil, err
}
case types.CIDTypeMetadataWebapp:
md = metadata.NewEmptyWebAppMetadata()
err = msgpack.Unmarshal(bytes, md)
if err != nil {
return nil, err
}
case types.CIDTypeDirectory:
md = metadata.NewEmptyDirectoryMetadata()
err = msgpack.Unmarshal(bytes, md)
if err != nil {
return nil, err
}
default:
return nil, errors.New("unsupported metadata format")
}
s.metadataCache.Put(hashStr, md)
return md, nil
}