diff --git a/marketplace/BUILD.bazel b/marketplace/BUILD.bazel index b43c2ef4bb..a39c3cc598 100644 --- a/marketplace/BUILD.bazel +++ b/marketplace/BUILD.bazel @@ -34,7 +34,6 @@ go_library( "//private/trust:go_default_library", "@com_connectrpc_connect//:go_default_library", "@com_github_golang_jwt_jwt//:go_default_library", - "@org_golang_google_protobuf//types/known/timestamppb:go_default_library", "@org_golang_x_crypto//bcrypt:go_default_library", ], ) diff --git a/marketplace/db/schema.sql b/marketplace/db/schema.sql index a971c9d063..1e4dc7d3a4 100644 --- a/marketplace/db/schema.sql +++ b/marketplace/db/schema.sql @@ -3,24 +3,33 @@ CREATE TABLE IF NOT EXISTS Assets( account_id INTEGER, isd_id INTEGER NOT NULL, as_id INTEGER NOT NULL, - bandwidth INTEGER NOT NULL, + bandwidth INTEGER NOT NULL CHECK (bandwidth > 0), bandwidth_min INTEGER NOT NULL, bandwidth_max INTEGER NOT NULL CHECK (bandwidth_max >= bandwidth_min), price INTEGER NOT NULL, time_granularity INTEGER NOT NULL, time_min_duration INTEGER NOT NULL, time_max_duration INTEGER NOT NULL, - starts_at TEXT NOT NULL, - stops_at TEXT NOT NULL, + starts_at INTEGER NOT NULL, + stops_at INTEGER NOT NULL, ingress INTEGER, egress INTEGER, state INTEGER NOT NULL DEFAULT 0, FOREIGN KEY (account_id) REFERENCES Accounts(id) ); -CREATE INDEX IF NOT EXISTS idx_asset_owner ON Assets(account_id); -CREATE INDEX IF NOT EXISTS idx_asset_validity ON Assets(starts_at, stops_at); -CREATE INDEX IF NOT EXISTS idx_assets_ia_validity_start ON Assets(isd_id, as_id, starts_at, stops_at); -CREATE INDEX IF NOT EXISTS idx_assets_ia_validity_end ON Assets(isd_id, as_id, stops_at, starts_at); +CREATE INDEX IF NOT EXISTS idx_asset_owner ON Assets(account_id, isd_id, as_id); +CREATE INDEX IF NOT EXISTS idx_assets_validity ON Assets(isd_id, as_id, stops_at); + +CREATE TABLE IF NOT EXISTS Asset_Segment( + asset_id INTEGER NOT NULL, + starts_at INTEGER NOT NULL, + stops_at INTEGER NOT NULL, + available INTEGER NOT NULL CHECK (available >= 0), + PRIMARY KEY (asset_id, starts_at), + CHECK (starts_at < stops_at), + FOREIGN KEY(asset_id) REFERENCES Assets(id) ON DELETE CASCADE +) WITHOUT ROWID; + CREATE TABLE IF NOT EXISTS Reservations( id INTEGER PRIMARY KEY AUTOINCREMENT, reservation_id INTEGER NOT NULL, @@ -31,8 +40,8 @@ CREATE TABLE IF NOT EXISTS Reservations( egress INTEGER NOT NULL, bandwidth INTEGER NOT NULL, bw_encoded INTEGER NOT NULL, - starts_at TEXT NOT NULL, - stops_at TEXT NOT NULL, + starts_at INTEGER NOT NULL, + stops_at INTEGER NOT NULL, key BLOB NOT NULL, FOREIGN KEY (account_id) REFERENCES Accounts(id) ); @@ -70,8 +79,8 @@ CREATE TABLE IF NOT EXISTS Redemption_Delegations( as_id INTEGER NOT NULL, res_id_limit_low INTEGER NOT NULL DEFAULT 0, res_id_limit_high INTEGER NOT NULL, - expiration TEXT NOT NULL CHECK (expiration <= paid_until), - paid_until TEXT NOT NULL, + expiration INTEGER NOT NULL CHECK (expiration <= paid_until), + paid_until INTEGER NOT NULL, key BLOB NOT NULL, encodings BLOB NOT NULL, PRIMARY KEY(isd_id, as_id) @@ -83,10 +92,10 @@ CREATE TABLE IF NOT EXISTS Asset_Events( ingress INTEGER, egress INTEGER, bandwidth INTEGER NOT NULL, - starts_at TEXT NOT NULL, - stops_at TEXT NOT NULL, + starts_at INTEGER NOT NULL, + stops_at INTEGER NOT NULL, price INTEGER NOT NULL, event_type INTEGER NOT NULL, - event_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')) + event_at INTEGER NOT NULL DEFAULT (unixepoch()) ); CREATE INDEX IF NOT EXISTS idx_asset_events ON Asset_Events(event_type, isd_id, as_id, stops_at, starts_at); \ No newline at end of file diff --git a/marketplace/db/sqlite.go b/marketplace/db/sqlite.go index ee292d9903..40f6c017fe 100644 --- a/marketplace/db/sqlite.go +++ b/marketplace/db/sqlite.go @@ -33,7 +33,11 @@ import ( type Repository interface { InsertAsset(ctx context.Context, a *DBAsset) (int64, error) InsertReservation(ctx context.Context, r *DBReservation) (int64, error) - Search(ctx context.Context, params *AssetQuery) ([]*DBAsset, error) + SearchOwnedAssets(ctx context.Context, params *OwnedAssetQuery) ([]*DBAsset, error) + SearchListedAssets(ctx context.Context, params *ListedAssetQuery) ([]*DBAsset, error) + LookupAsset(ctx context.Context, id int64, ia addr.IA) (*DBAsset, error) + LookupAssets(ctx context.Context, ia addr.IA) ([]*DBAsset, error) + LookupAllAssetSegments(ctx context.Context, id int64) ([]*DBAssetSegment, error) FetchReservations(ctx context.Context, params *ReservationQuery) ([]*DBReservation, error) GetUser(ctx context.Context, id int64) (*DBUser, error) GetUserByName(ctx context.Context, name string) (*DBUser, error) @@ -64,6 +68,13 @@ type Repository interface { RemoveAsset(ctx context.Context, assetID int64) error RegisterAssetEvent(ctx context.Context, a *DBAsset, eventType AssetEventType) (int64, error) TransitionAsset(ctx context.Context, assetID int64, accountID *int64, from AssetState, to AssetState) (*DBAsset, error) + InsertAssetSegment(ctx context.Context, asset_id int64, from int64, to int64, bw uint32) (int64, error) + CheckAssetAvailability(ctx context.Context, assetId int64, startsAt int64, stopsAt int64, bw uint32) (*DBAsset, error) + SplitAssetAt(ctx context.Context, assetId int64, point int64) error + MergeAssetAt(ctx context.Context, assetId int64, point int64) error + IncreaseAssetSegmentsBandwidth(ctx context.Context, assetId int64, delta int64) error + UpdateAsset(ctx context.Context, a *DBAsset) (int64, error) + BuyAsset(ctx context.Context, startsAt int64, stopsAt int64, bw uint32, accountId int64, dbAsset *DBAsset) (*DBAsset, error) } type MarketplaceDB interface { @@ -152,12 +163,13 @@ type rowScanner interface { Scan(dest ...any) error } -func scanAsset(row rowScanner) (*DBAsset, error) { +func scanAssetWithAvailableBandwidthAndUsability(row rowScanner) (*DBAsset, error) { asset := &DBAsset{} - var startsAt string - var stopsAt string var isd uint16 var as uint64 + var usableFrom int64 + var usableTo int64 + var availableBW uint32 if err := row.Scan( &asset.ID, &asset.AccountId, @@ -170,10 +182,13 @@ func scanAsset(row rowScanner) (*DBAsset, error) { &asset.TimeGranularity, &asset.TimeMinDuration, &asset.TimeMaxDuration, - &startsAt, - &stopsAt, + &asset.StartsAt, + &asset.StopsAt, &asset.IfIdIngress, &asset.IfIdEgress, + &usableFrom, + &usableTo, + &availableBW, ); err != nil { return nil, serrors.Wrap("Error reading DB response", err) } @@ -182,11 +197,71 @@ func scanAsset(row rowScanner) (*DBAsset, error) { if err != nil { return nil, err } - asset.StartAt, err = time.Parse(time.RFC3339, startsAt) + asset.Bandwidth = availableBW + asset.StartsAt = usableFrom + asset.StopsAt = usableTo + return asset, nil +} + +func scanAssetWithAvailableBandwidth(row rowScanner) (*DBAsset, error) { + asset := &DBAsset{} + var isd uint16 + var as uint64 + var availableBW uint32 + if err := row.Scan( + &asset.ID, + &asset.AccountId, + &isd, + &as, + &asset.Bandwidth, + &asset.BandwidthMin, + &asset.BandwidthMax, + &asset.Price, + &asset.TimeGranularity, + &asset.TimeMinDuration, + &asset.TimeMaxDuration, + &asset.StartsAt, + &asset.StopsAt, + &asset.IfIdIngress, + &asset.IfIdEgress, + &availableBW, + ); err != nil { + return nil, serrors.Wrap("Error reading DB response", err) + } + var err error + asset.IA, err = addr.IAFrom(addr.ISD(isd), addr.AS(as)) if err != nil { return nil, err } - asset.StopsAt, err = time.Parse(time.RFC3339, stopsAt) + asset.Bandwidth = availableBW + return asset, nil +} + +func scanAsset(row rowScanner) (*DBAsset, error) { + asset := &DBAsset{} + var isd uint16 + var as uint64 + if err := row.Scan( + &asset.ID, + &asset.AccountId, + &isd, + &as, + &asset.Bandwidth, + &asset.BandwidthMin, + &asset.BandwidthMax, + &asset.Price, + &asset.TimeGranularity, + &asset.TimeMinDuration, + &asset.TimeMaxDuration, + &asset.StartsAt, + &asset.StopsAt, + &asset.IfIdIngress, + &asset.IfIdEgress, + ); err != nil { + return nil, serrors.Wrap("Error reading DB response", err) + } + var err error + asset.IA, err = addr.IAFrom(addr.ISD(isd), addr.AS(as)) if err != nil { return nil, err } @@ -195,8 +270,6 @@ func scanAsset(row rowScanner) (*DBAsset, error) { func scanReservation(row rowScanner) (*DBReservation, error) { reservation := &DBReservation{} - var startsAt string - var stopsAt string var isd uint16 var as uint64 if err := row.Scan( @@ -209,8 +282,8 @@ func scanReservation(row rowScanner) (*DBReservation, error) { &reservation.Egress, &reservation.Bandwidth, &reservation.EncodedBandwidth, - &startsAt, - &stopsAt, + &reservation.StartsAt, + &reservation.StopsAt, &reservation.Key, ); err != nil { return nil, serrors.Wrap("Error reading DB response", err) @@ -220,14 +293,6 @@ func scanReservation(row rowScanner) (*DBReservation, error) { if err != nil { return nil, err } - reservation.StartsAt, err = time.Parse(time.RFC3339, startsAt) - if err != nil { - return nil, err - } - reservation.StopsAt, err = time.Parse(time.RFC3339, stopsAt) - if err != nil { - return nil, err - } return reservation, nil } @@ -303,7 +368,7 @@ func (e *executor) FindRedemptionDelegations( } stmt := `SELECT isd_id, as_id, res_id_limit_low, res_id_limit_high, expiration, paid_until, key, encodings FROM Redemption_Delegations WHERE expiration >= ?` - args := []any{time.Now().UTC().Format(time.RFC3339)} + args := []any{time.Now().Unix()} rows, err := e.read.QueryContext(ctx, stmt, args...) if err != nil { return nil, serrors.New("Error looking up assets", "err", err, "q", stmt) @@ -312,8 +377,6 @@ func (e *executor) FindRedemptionDelegations( var res []*RedemptionDelegation for rows.Next() { a := &RedemptionDelegation{} - var expirationString string - var paidUntilString string var isd uint16 var as uint64 err = rows.Scan( @@ -321,21 +384,13 @@ func (e *executor) FindRedemptionDelegations( &as, &a.ResIdLow, &a.ResIdHigh, - &expirationString, - &paidUntilString, + &a.Expiration, + &a.PaidUntil, &a.Key, &a.Encodings) if err != nil { return nil, serrors.Wrap("Error reading DB response", err) } - a.Expiration, err = time.Parse(time.RFC3339, expirationString) - if err != nil { - return nil, err - } - a.PaidUntil, err = time.Parse(time.RFC3339, paidUntilString) - if err != nil { - return nil, err - } a.IA, err = addr.IAFrom(addr.ISD(isd), addr.AS(as)) res = append(res, a) } @@ -352,7 +407,7 @@ func (e *executor) FindRedemptionDelegation( stmt := `SELECT isd_id, as_id, res_id_limit_low, res_id_limit_high, expiration, paid_until, key, encodings FROM Redemption_Delegations WHERE expiration >= ? AND isd_id = ? AND as_id = ?` - rows, err := e.read.QueryContext(ctx, stmt, time.Now().Format(time.RFC3339), ia.ISD(), ia.AS()) + rows, err := e.read.QueryContext(ctx, stmt, time.Now().Unix(), ia.ISD(), ia.AS()) if err != nil { return nil, serrors.New("Error looking up assets", "err", err, "q", stmt) } @@ -361,8 +416,6 @@ func (e *executor) FindRedemptionDelegation( return nil, nil } a := &RedemptionDelegation{} - var expirationString string - var paidUntilString string var isd uint16 var as uint64 err = rows.Scan( @@ -370,21 +423,13 @@ func (e *executor) FindRedemptionDelegation( &as, &a.ResIdLow, &a.ResIdHigh, - &expirationString, - &paidUntilString, + &a.Expiration, + &a.PaidUntil, &a.Key, &a.Encodings) if err != nil { return nil, serrors.Wrap("Error reading DB response", err) } - a.Expiration, err = time.Parse(time.RFC3339, expirationString) - if err != nil { - return nil, err - } - a.PaidUntil, err = time.Parse(time.RFC3339, paidUntilString) - if err != nil { - return nil, err - } a.IA, err = addr.IAFrom(addr.ISD(isd), addr.AS(as)) return a, nil } @@ -490,8 +535,8 @@ func (e *executor) CreateOrUpdateRedemptionDelegations( r.IA.AS(), r.ResIdLow, r.ResIdHigh, - r.Expiration.UTC().Format(time.RFC3339), - r.PaidUntil.UTC().Format(time.RFC3339), + r.Expiration, + r.PaidUntil, r.Key, r.Encodings) if err != nil { @@ -516,20 +561,10 @@ func (e *executor) FindUsedReservations( var res []*UsedReservation for rows.Next() { a := &UsedReservation{} - var startsAtString string - var stopsAtString string - err = rows.Scan(&a.Id, &startsAtString, &stopsAtString) + err = rows.Scan(&a.Id, &a.StartsAt, &a.StopsAt) if err != nil { return nil, serrors.Wrap("Error reading DB response", err) } - a.StartsAt, err = time.Parse(time.RFC3339, startsAtString) - if err != nil { - return nil, err - } - a.StopsAt, err = time.Parse(time.RFC3339, stopsAtString) - if err != nil { - return nil, err - } res = append(res, a) } return res, nil @@ -544,7 +579,7 @@ func (e *executor) buildUsedReservationsQuery(params *UsedReservationsQuery) (st params.IA.AS(), params.Limit_low, params.Limit_high, - time.Now().UTC().Format(time.RFC3339), + time.Now().Unix(), } return query, args } @@ -562,20 +597,10 @@ func (e *executor) SearchAssetsForStatistics(ctx context.Context, params *Statis var publishedAssets []*DBStat for publishRows.Next() { a := &DBStat{} - var startsAtString string - var stopsAtString string - err = publishRows.Scan(&a.Bandwidth, &a.Price, &startsAtString, &stopsAtString) + err = publishRows.Scan(&a.Bandwidth, &a.Price, &a.StartsAt, &a.StopsAt) if err != nil { return nil, nil, serrors.Wrap("Error reading DB response", err) } - a.StartsAt, err = time.Parse(time.RFC3339, startsAtString) - if err != nil { - return nil, nil, err - } - a.StopsAt, err = time.Parse(time.RFC3339, stopsAtString) - if err != nil { - return nil, nil, err - } publishedAssets = append(publishedAssets, a) } boughtStmt, args := e.buildAssetEventsQuery(params, AssetBought) @@ -587,20 +612,10 @@ func (e *executor) SearchAssetsForStatistics(ctx context.Context, params *Statis var boughtAssets []*DBStat for boughtRows.Next() { a := &DBStat{} - var startsAtString string - var stopsAtString string - err = boughtRows.Scan(&a.Bandwidth, &a.Price, &startsAtString, &stopsAtString) + err = boughtRows.Scan(&a.Bandwidth, &a.Price, &a.StartsAt, &a.StopsAt) if err != nil { return nil, nil, serrors.Wrap("Error reading DB response", err) } - a.StartsAt, err = time.Parse(time.RFC3339, startsAtString) - if err != nil { - return nil, nil, err - } - a.StopsAt, err = time.Parse(time.RFC3339, stopsAtString) - if err != nil { - return nil, nil, err - } boughtAssets = append(boughtAssets, a) } return publishedAssets, boughtAssets, nil @@ -634,18 +649,26 @@ func (e *executor) RegisterAssetEvent(ctx context.Context, a *DBAsset, eventType inst := `INSERT INTO Asset_Events (event_type, isd_id, as_id, ingress, egress, bandwidth, starts_at, stops_at, price) VALUES(?,?,?,?,?,?,?,?,?)` res, err := e.write.ExecContext(ctx, inst, eventType, a.IA.ISD(), a.IA.AS(), a.IfIdIngress, a.IfIdEgress, a.Bandwidth, - a.StartAt.UTC().Format(time.RFC3339), a.StopsAt.UTC().Format(time.RFC3339), a.Price) + a.StartsAt, a.StopsAt, a.Price) if err != nil { return 0, err } return res.LastInsertId() } -func (e *executor) Search(ctx context.Context, params *AssetQuery) ([]*DBAsset, error) { +func (e *executor) SearchOwnedAssets(ctx context.Context, params *OwnedAssetQuery) ([]*DBAsset, error) { if e.read == nil { return nil, serrors.New("No database open") } - stmt, args := e.buildSearchQuery(params) + var stmt string + var args []any + if params.ValidityMode == FullOverlap { + stmt, args = e.buildSearchOwnedAssetsFullOverlapQuery(params) + } else if params.ValidityMode == PartialOverlap { + stmt, args = e.buildSearchOwnedAssetsPartialOverlapQuery(params) + } else { + return nil, serrors.New("invalid validity mode") + } rows, err := e.read.QueryContext(ctx, stmt, args...) if err != nil { return nil, serrors.New("Error looking up assets", "err", err, "q", stmt) @@ -662,23 +685,268 @@ func (e *executor) Search(ctx context.Context, params *AssetQuery) ([]*DBAsset, return res, nil } -func (e *executor) buildSearchQuery(params *AssetQuery) (string, []any) { +func (e *executor) LookupAsset(ctx context.Context, id int64, ia addr.IA) (*DBAsset, error) { + if e.read == nil { + return nil, serrors.New("No database open") + } + q := `SELECT ` + assetColumnsWithAlias + ` FROM Assets a WHERE a.ID = ? AND a.isd_id = ? AND a.as_id = ?` + rows, err := e.read.QueryContext(ctx, q, id, ia.ISD(), ia.AS()) + if err != nil { + return nil, err + } + if !rows.Next() { + return nil, serrors.New("asset not found") + } + return scanAsset(rows) +} + +func (e *executor) LookupAssets(ctx context.Context, ia addr.IA) ([]*DBAsset, error) { + if e.read == nil { + return nil, serrors.New("No database open") + } + q := `SELECT ` + assetColumnsWithAlias + ` FROM Assets a WHERE a.isd_id = ? AND a.as_id = ? AND a.account_id IS NULL` + rows, err := e.read.QueryContext(ctx, q, ia.ISD(), ia.AS()) + if err != nil { + return nil, err + } + res := []*DBAsset{} + for rows.Next() { + asset, err := scanAsset(rows) + if err != nil { + return nil, err + } + res = append(res, asset) + } + return res, nil +} + +func (e *executor) LookupAllAssetSegments(ctx context.Context, id int64) ([]*DBAssetSegment, error) { + if e.read == nil { + return nil, serrors.New("No database open") + } + q := `SELECT starts_at, stops_at, available FROM Asset_Segment WHERE asset_id = ?` + rows, err := e.read.QueryContext(ctx, q, id) + if err != nil { + return nil, err + } + res := []*DBAssetSegment{} + for rows.Next() { + seg := &DBAssetSegment{} + err = rows.Scan(&seg.StartsAt, &seg.StopsAt, &seg.Available) + if err != nil { + return nil, err + } + res = append(res, seg) + } + return res, nil +} + +func (e *executor) SearchListedAssets(ctx context.Context, params *ListedAssetQuery) ([]*DBAsset, error) { + if e.read == nil { + return nil, serrors.New("No database open") + } + var stmt string + var args []any + var res []*DBAsset + if params.ValidityMode == FullOverlap { + stmt, args = e.buildSearchListedAssetsFullOverlapQuery(params) + rows, err := e.read.QueryContext(ctx, stmt, args...) + if err != nil { + return nil, serrors.New("Error looking up assets", "err", err, "q", stmt) + } + defer rows.Close() + + for rows.Next() { + a, err := scanAssetWithAvailableBandwidth(rows) + if err != nil { + return nil, err + } + res = append(res, a) + } + } else if params.ValidityMode == PartialOverlap { + stmt, args = e.buildSearchListedAssetsPartialOverlapQuery(params) + rows, err := e.read.QueryContext(ctx, stmt, args...) + if err != nil { + return nil, serrors.New("Error looking up assets", "err", err, "q", stmt) + } + defer rows.Close() + for rows.Next() { + a, err := scanAssetWithAvailableBandwidthAndUsability(rows) + if err != nil { + return nil, err + } + res = append(res, a) + } + } else { + return nil, serrors.New("invalid validity mode") + } + + return res, nil +} + +func (e *executor) buildSearchListedAssetsPartialOverlapQuery(params *ListedAssetQuery) (string, []any) { + var args []any + where := []string{} + queryStart := []string{` + WITH candidates AS ( + SELECT ` + assetColumnsWithAlias + ` + FROM Assets a + WHERE a.isd_id = ? AND a.as_id = ? + AND a.stops_at > ? AND a.starts_at < ? + AND a.account_id IS NULL + AND a.bandwidth >= ? + `} + queryEnd := ` + ), + hits AS ( + SELECT c.id AS asset_id, + MAX(g.starts_at, ?) AS f, + MIN(g.stops_at, ?) AS t, + g.available + FROM candidates c + JOIN Asset_Segment g + ON g.asset_id = c.id AND g.starts_at < ? AND g.stops_at > ? + WHERE g.available >= ? + ), + marked AS ( + SELECT *, + CASE WHEN f = LAG(t) OVER (PARTITION BY asset_id ORDER BY f) + THEN 0 ELSE 1 END AS new_island + FROM hits + ), + islands AS ( + SELECT *, + SUM(new_island) OVER (PARTITION BY asset_id ORDER BY f) AS island + FROM marked + ) + SELECT c.id, c.account_id, c.isd_id, c.as_id, c.bandwidth, c.bandwidth_min, c.bandwidth_max, + c.price, c.time_granularity, c.time_min_duration, c.time_max_duration, c.starts_at, c.stops_at, + c.ingress, c.egress, + MIN(i.f) AS usable_from, + MAX(i.t) AS usable_to, + MIN(i.available) AS max_buyable_bandwidth + FROM islands i + JOIN candidates c ON c.id = i.asset_id + GROUP BY i.asset_id, i.island + ORDER BY c.isd_id, c.as_id, c.ingress, c.egress, c.id, usable_from` + args = append(args, params.IA.ISD(), params.IA.AS(), params.StartsAt, params.StopsAt, params.MinRequiredBandwidth) + where = append(where, "(a.state = ?)") + args = append(args, AssetStateAvailable) + if params.Price != nil { + where = append(where, "(a.price<=?)") + args = append(args, *params.Price) + } + if params.Ingress != nil { + where = append(where, "(a.ingress=?)") + args = append(args, *params.Ingress) + } + if params.Egress != nil { + where = append(where, "(a.egress=?)") + args = append(args, *params.Egress) + } + args = append(args, params.StartsAt, params.StopsAt, params.StopsAt, params.StartsAt, params.MinRequiredBandwidth) + query := append(queryStart, fmt.Sprintf("AND %s", strings.Join(where, "AND\n")), queryEnd) + return strings.Join(query, "\n"), args +} + +func (e *executor) buildSearchListedAssetsFullOverlapQuery(params *ListedAssetQuery) (string, []any) { + var args []any + where := []string{} + queryStart := []string{` + + SELECT ` + assetColumnsWithAlias + `, MIN(g.available) AS max_buyable_bandwidth + FROM Assets a + JOIN Asset_Segment g + ON g.asset_id = a.id AND g.starts_at < ? AND g.stops_at > ? + WHERE a.isd_id = ? AND a.as_id = ? + AND a.starts_at <= ? AND a.stops_at >= ? + AND a.account_id IS NULL + AND a.bandwidth >= ? + `, + } + queryEnd := ` + GROUP BY a.id + HAVING MIN(g.available) >= ? + ORDER BY a.id + ` + args = append(args, params.StopsAt, params.StartsAt, params.IA.ISD(), params.IA.AS(), params.StartsAt, + params.StopsAt, params.MinRequiredBandwidth) + + where = append(where, "(a.state = ?)") + args = append(args, AssetStateAvailable) + if params.Price != nil { + where = append(where, "(a.price<=?)") + args = append(args, *params.Price) + } + if params.Ingress != nil { + where = append(where, "(a.ingress=?)") + args = append(args, *params.Ingress) + } + if params.Egress != nil { + where = append(where, "(a.egress=?)") + args = append(args, *params.Egress) + } + args = append(args, params.MinRequiredBandwidth) + query := append(queryStart, fmt.Sprintf("AND %s", strings.Join(where, "AND\n")), queryEnd) + return strings.Join(query, "\n"), args +} + +func (e *executor) buildSearchOwnedAssetsPartialOverlapQuery(params *OwnedAssetQuery) (string, []any) { var args []any where := []string{} query := []string{ "SELECT " + assetColumnsWithAlias + " FROM Assets a", } - if params.AccountId == nil { - where = append(where, "(a.account_id IS NULL)") - } else { - // The JOIN clause will be the first parametrized entry. Fill up the arguments for it. - query = append(query, "JOIN Accounts owner ON a.account_id = owner.id JOIN "+ - "Accounts current ON current.id = ?") - args = append(args, *params.AccountId) - // Parenthesized as a whole: AND to all where[i] binds tighter than OR. - where = append(where, "( (current.scope = '' AND owner.user_id = current.user_id) OR "+ - "(current.scope != '' AND owner.id = current.id) )") + query = append(query, "JOIN Accounts owner ON a.account_id = owner.id JOIN "+ + "Accounts current ON current.id = ?") + args = append(args, *params.AccountId) + where = append(where, "((current.scope = '' AND owner.user_id = current.user_id) OR "+ + "(current.scope != '' AND owner.id = current.id))") + where = append(where, "(a.state = ?)") + args = append(args, AssetStateAvailable) + if params.IA != nil { + where = append(where, "(a.isd_id=?) AND (a.as_id=?)") + args = append(args, int64(params.IA.ISD()), int64(params.IA.AS())) + } + if params.StartsAt != nil { + where = append(where, "(a.starts_at>=?)") + args = append(args, *params.StartsAt) } + if params.StopsAt != nil { + where = append(where, "(a.stops_at<=?)") + args = append(args, *params.StopsAt) + } + if params.Price != nil { + where = append(where, "(a.price<=?)") + args = append(args, *params.Price) + } + if params.MinRequiredBandwidth != nil { + where = append(where, "(a.bandwidth>=?)") + args = append(args, *params.MinRequiredBandwidth) + } + if params.Ingress != nil { + where = append(where, "(a.ingress=?)") + args = append(args, *params.Ingress) + } + if params.Egress != nil { + where = append(where, "(a.egress=?)") + args = append(args, *params.Egress) + } + query = append(query, fmt.Sprintf("WHERE %s", strings.Join(where, "AND\n"))) + return strings.Join(query, "\n"), args +} + +func (e *executor) buildSearchOwnedAssetsFullOverlapQuery(params *OwnedAssetQuery) (string, []any) { + var args []any + where := []string{} + query := []string{ + "SELECT " + assetColumnsWithAlias + " FROM Assets a", + } + query = append(query, "JOIN Accounts owner ON a.account_id = owner.id JOIN "+ + "Accounts current ON current.id = ?") + args = append(args, *params.AccountId) + where = append(where, "((current.scope = '' AND owner.user_id = current.user_id) OR "+ + "(current.scope != '' AND owner.id = current.id))") where = append(where, "(a.state = ?)") args = append(args, AssetStateAvailable) if params.IA != nil { @@ -710,9 +978,6 @@ func (e *executor) buildSearchQuery(params *AssetQuery) (string, []any) { args = append(args, *params.Egress) } query = append(query, fmt.Sprintf("WHERE %s", strings.Join(where, "AND\n"))) - query = append(query, "ORDER BY LENGTH(a.id) ASC, a.id ASC") - query = append(query, "LIMIT ? OFFSET ?") - args = append(args, params.PageSize, params.Page*params.PageSize) return strings.Join(query, "\n"), args } @@ -766,8 +1031,8 @@ func (e *executor) InsertReservation(ctx context.Context, r *DBReservation) (int r.Egress, r.Bandwidth, r.EncodedBandwidth, - r.StartsAt.UTC().Format(time.RFC3339), - r.StopsAt.UTC().Format(time.RFC3339), + r.StartsAt, + r.StopsAt, r.Key, r.AccountId) if err != nil { @@ -797,8 +1062,8 @@ func (e *executor) InsertAsset(ctx context.Context, a *DBAsset) (int64, error) { a.TimeGranularity, a.TimeMinDuration, a.TimeMaxDuration, - a.StartAt.UTC().Format(time.RFC3339), - a.StopsAt.UTC().Format(time.RFC3339), + a.StartsAt, + a.StopsAt, a.IfIdIngress, a.IfIdEgress, a.AccountId) @@ -815,8 +1080,8 @@ func (e *executor) InsertAsset(ctx context.Context, a *DBAsset) (int64, error) { a.TimeGranularity, a.TimeMinDuration, a.TimeMaxDuration, - a.StartAt.UTC().Format(time.RFC3339), - a.StopsAt.UTC().Format(time.RFC3339), + a.StartsAt, + a.StopsAt, a.IfIdIngress, a.IfIdEgress) } @@ -1147,3 +1412,222 @@ func (e *executor) AssignReservation( } return res.RowsAffected() } + +func (e *executor) InsertAssetSegment(ctx context.Context, asset_id int64, from int64, to int64, bw uint32) (int64, error) { + if e.write == nil { + return 0, serrors.New("No database open") + } + inst := `INSERT INTO Asset_Segment(asset_id, starts_at, stops_at, available) VALUES (?,?,?,?)` + res, err := e.write.ExecContext(ctx, inst, asset_id, from, to, bw) + if err != nil { + return 0, err + } + return res.LastInsertId() +} + +func (e *executor) IncreaseAssetSegmentsBandwidth(ctx context.Context, assetId int64, delta int64) error { + if e.write == nil { + return serrors.New("No database open") + } + inst := ` + UPDATE Asset_Segment SET available = available + ? + WHERE asset_id = ?` + _, err := e.write.ExecContext(ctx, inst, delta, assetId) + if err != nil { + return err + } + return nil +} + +func (e *executor) UpdateAsset(ctx context.Context, a *DBAsset) (int64, error) { + if e.write == nil { + return 0, serrors.New("No database open") + } + inst := ` + UPDATE Assets + SET bandwidth = ?, bandwidth_min = ?, bandwidth_max = ?, price = ?, time_granularity = ?, + time_min_duration = ?, time_max_duration = ?, starts_at = ?, stops_at = ? + WHERE id = ? AND state = ? AND account_ID IS NULL` + res, err := e.write.ExecContext(ctx, inst, a.Bandwidth, a.BandwidthMin, a.BandwidthMax, + a.Price, a.TimeGranularity, a.TimeMinDuration, a.TimeMaxDuration, a.StartsAt, + a.StopsAt, a.ID, AssetStateAvailable) + if err != nil { + return 0, err + } + return res.RowsAffected() +} + +func (e *executor) MergeAssetAt(ctx context.Context, assetId int64, point int64) error { + if e.write == nil { + return serrors.New("No database open") + } + updateInst := ` + UPDATE Asset_Segment AS l + SET stops_at = r.stops_at + FROM Asset_Segment AS r + WHERE l.asset_id = ? AND l.stops_at = ? + AND r.asset_id = ? AND r.starts_at = ? + AND r.available = l.available + ` + res, err := e.write.ExecContext(ctx, updateInst, assetId, point, assetId, point) + if err != nil { + return fmt.Errorf("merge at %d (update): %w", point, err) + } + if n, _ := res.RowsAffected(); n == 1 { + deleteInst := `DELETE FROM Asset_Segment WHERE asset_id = ? AND starts_at = ?` + if _, err := e.write.ExecContext(ctx, deleteInst, assetId, point); err != nil { + return fmt.Errorf("merge at %d (delete): %w", point, err) + } + } + return nil +} + +func (e *executor) SplitAssetAt(ctx context.Context, assetId int64, point int64) error { + if e.write == nil { + return serrors.New("No database open") + } + insertInst := ` + INSERT INTO Asset_segment (asset_id, starts_at, stops_at, available) + SELECT asset_id, ?, stops_at, available + FROM Asset_segment + WHERE asset_id = ? AND starts_at < ? AND stops_at > ?` + _, err := e.write.ExecContext(ctx, insertInst, point, assetId, point, point) + if err != nil { + return err + } + updateInst := ` + UPDATE Asset_segment + SET stops_at = ? + WHERE asset_id = ? AND starts_at < ? AND stops_at > ? + ` + _, err = e.write.ExecContext(ctx, updateInst, point, assetId, point, point) + if err != nil { + return err + } + return nil +} + +func (e *executor) CheckAssetAvailability( + ctx context.Context, + assetId int64, + startsAt int64, + stopsAt int64, + bw uint32, +) (*DBAsset, error) { + if e.read == nil { + return nil, serrors.New("No database open") + } + inst := ` + SELECT ` + assetColumnsWithAlias + ` FROM Assets a + WHERE a.id = ? + AND a.account_ID IS NULL + AND a.state = ? + AND a.starts_at <= ? AND a.stops_at >= ? + AND NOT EXISTS ( + SELECT 1 FROM Asset_Segment g + WHERE g.asset_id = a.id + AND g.starts_at < ? AND g.stops_at > ? + AND g.available < ?); + ` + rows, err := e.read.QueryContext(ctx, inst, assetId, AssetStateAvailable, startsAt, stopsAt, stopsAt, startsAt, bw) + if err != nil { + return nil, err + } + if !rows.Next() { + return nil, serrors.New("asset not found") + } + asset, err := scanAsset(rows) + if err != nil { + return nil, err + } + return asset, nil +} + +func (e *executor) BuyFromAssetSegment( + ctx context.Context, + assetId int64, + startsAt int64, + stopsAt int64, + bw uint32, + accountId int64, + asset *DBAsset, +) (*DBAsset, error) { + decreaseAvailabilityInst := ` + UPDATE asset_segment + SET available = available - ? + WHERE asset_id = ? AND starts_at >= ? AND stops_at <= ?` + _, err := e.write.ExecContext(ctx, decreaseAvailabilityInst, bw, assetId, startsAt, stopsAt) + if err != nil { + return nil, err + } + userAsset := &DBAsset{ + AccountId: sql.NullInt64{ + Int64: accountId, + Valid: true, + }, + IA: asset.IA, + Bandwidth: bw, + StartsAt: startsAt, + StopsAt: stopsAt, + BandwidthMin: asset.BandwidthMin, + BandwidthMax: asset.BandwidthMax, + Price: asset.Price, + TimeGranularity: asset.TimeGranularity, + TimeMinDuration: asset.TimeMinDuration, + TimeMaxDuration: asset.TimeMaxDuration, + IfIdIngress: asset.IfIdIngress, + IfIdEgress: asset.IfIdEgress, + } + newId, err := e.InsertAsset(ctx, userAsset) + if err != nil { + return nil, err + } + userAsset.ID = newId + return userAsset, nil +} + +// Buys a single asset. Availability must be checked in advance in same transaction. +func (e *executor) BuyAsset( + ctx context.Context, + startsAt int64, + stopsAt int64, + bw uint32, + accountId int64, + dbAsset *DBAsset, +) (*DBAsset, error) { + if e.write == nil { + return nil, serrors.New("No database open") + } + decreaseAvailabilityInst := ` + UPDATE asset_segment + SET available = available - ? + WHERE asset_id = ? AND starts_at >= ? AND stops_at <= ?` + _, err := e.write.ExecContext(ctx, decreaseAvailabilityInst, bw, dbAsset.ID, startsAt, stopsAt) + if err != nil { + return nil, err + } + userAsset := &DBAsset{ + AccountId: sql.NullInt64{ + Int64: accountId, + Valid: true, + }, + IA: dbAsset.IA, + Bandwidth: bw, + StartsAt: startsAt, + StopsAt: stopsAt, + BandwidthMin: dbAsset.BandwidthMin, + BandwidthMax: dbAsset.BandwidthMax, + Price: dbAsset.Price, + TimeGranularity: dbAsset.TimeGranularity, + TimeMinDuration: dbAsset.TimeMinDuration, + TimeMaxDuration: dbAsset.TimeMaxDuration, + IfIdIngress: dbAsset.IfIdIngress, + IfIdEgress: dbAsset.IfIdEgress, + } + newId, err := e.InsertAsset(ctx, userAsset) + if err != nil { + return nil, err + } + userAsset.ID = newId + return userAsset, nil +} diff --git a/marketplace/db/sqlite_test.go b/marketplace/db/sqlite_test.go index 7b11494d1d..2303cf2a43 100644 --- a/marketplace/db/sqlite_test.go +++ b/marketplace/db/sqlite_test.go @@ -47,11 +47,23 @@ func testAsset() *DBAsset { Price: 5, TimeGranularity: 60, TimeMinDuration: 60, - StartAt: start, - StopsAt: start.Add(time.Hour), + StartsAt: start.Unix(), + StopsAt: start.Add(time.Hour).Unix(), } } +func insertAsset(ctx context.Context, repo Repository, asset *DBAsset) (int64, error) { + assetId, err := repo.InsertAsset(ctx, asset) + if err != nil { + return 0, err + } + if !asset.AccountId.Valid { + _, err = repo.InsertAssetSegment(ctx, assetId, + asset.StartsAt, asset.StopsAt, asset.Bandwidth) + } + return assetId, err +} + // TestAssetIDInt64 rejects identifiers that cannot be represented by SQLite. func TestAssetIDInt64(t *testing.T) { id, err := AssetID(42).Int64() @@ -70,18 +82,23 @@ func TestWithTxCommitsAndRollsBack(t *testing.T) { // A successful callback commits its insert. err := backend.WithTx(ctx, func(repository Repository) error { - _, err := repository.InsertAsset(ctx, testAsset()) + _, err := insertAsset(ctx, repository, testAsset()) return err }) require.NoError(t, err) - assets, err := backend.Search(ctx, &AssetQuery{PageSize: 64}) + assets, err := backend.SearchListedAssets(ctx, &ListedAssetQuery{ + IA: addr.MustParseIA("1-ff00:0:110"), + StartsAt: 0, + StopsAt: math.MaxInt64, + ValidityMode: PartialOverlap, + }) require.NoError(t, err) require.Len(t, assets, 1) // Rolls back the insert and preserves the error. rollbackErr := errors.New("rollback") err = backend.WithTx(ctx, func(repository Repository) error { - _, err := repository.InsertAsset(ctx, testAsset()) + _, err := insertAsset(ctx, repository, testAsset()) if err != nil { return err } @@ -90,7 +107,12 @@ func TestWithTxCommitsAndRollsBack(t *testing.T) { require.ErrorIs(t, err, rollbackErr) // Only the asset written by the committed transaction is visible. - assets, err = backend.Search(ctx, &AssetQuery{PageSize: 64}) + assets, err = backend.SearchListedAssets(ctx, &ListedAssetQuery{ + IA: addr.MustParseIA("1-ff00:0:110"), + StartsAt: 0, + StopsAt: math.MaxInt64, + ValidityMode: PartialOverlap, + }) require.NoError(t, err) require.Len(t, assets, 1) } @@ -102,7 +124,7 @@ func TestAssetTransitionAndRowScanners(t *testing.T) { backend := newTestBackend(t) asset := testAsset() - assetID, err := backend.InsertAsset(ctx, asset) + assetID, err := insertAsset(ctx, backend, testAsset()) require.NoError(t, err) // Listing an asset without an owner lets the marketplace check it out. transitioned, err := backend.TransitionAsset(ctx, assetID, nil, @@ -113,7 +135,12 @@ func TestAssetTransitionAndRowScanners(t *testing.T) { require.False(t, transitioned.AccountId.Valid) // Checked-out assets are no longer returned by the available-asset search. - assets, err := backend.Search(ctx, &AssetQuery{PageSize: 64}) + assets, err := backend.SearchListedAssets(ctx, &ListedAssetQuery{ + IA: addr.MustParseIA("1-ff00:0:110"), + StartsAt: 0, + StopsAt: math.MaxInt64, + ValidityMode: PartialOverlap, + }) require.NoError(t, err) require.Empty(t, assets) @@ -130,7 +157,7 @@ func TestAssetTransitionAndRowScanners(t *testing.T) { Egress: 2, Bandwidth: 100, EncodedBandwidth: 5, - StartsAt: asset.StartAt, + StartsAt: asset.StartsAt, StopsAt: asset.StopsAt, Key: []byte("key"), } @@ -193,11 +220,11 @@ func TestSearchOwnedAssets(t *testing.T) { owned := testAsset() owned.AccountId = sql.NullInt64{Int64: alicesAccountID, Valid: true} - ownedID, err := backend.InsertAsset(ctx, owned) + ownedID, err := insertAsset(ctx, backend, owned) require.NoError(t, err) // An asset nobody owns, and one owned by somebody else. - notOwnedAssetID, err := backend.InsertAsset(ctx, testAsset()) + notOwnedAssetID, err := insertAsset(ctx, backend, testAsset()) require.NoError(t, err) bobUserID, err := backend.CreateUser(ctx, &DBUser{Name: "bob", PasswordHash: "hash"}) require.NoError(t, err) @@ -205,38 +232,43 @@ func TestSearchOwnedAssets(t *testing.T) { require.NoError(t, err) otherAsset := testAsset() otherAsset.AccountId = sql.NullInt64{Int64: bobsAccountID, Valid: true} - _, err = backend.InsertAsset(ctx, otherAsset) + _, err = insertAsset(ctx, backend, otherAsset) require.NoError(t, err) - assets, err := backend.Search(ctx, &AssetQuery{AccountId: &alicesAccountID, PageSize: 64}) + assets, err := backend.SearchOwnedAssets(ctx, &OwnedAssetQuery{ + AccountId: &alicesAccountID, + }) require.NoError(t, err) require.Len(t, assets, 1) require.Equal(t, ownedID, assets[0].ID) // The validity filters still apply to the assets of the account. - startsAt := owned.StartAt.UTC().Format(time.RFC3339) - stopsAt := owned.StopsAt.UTC().Format(time.RFC3339) - assets, err = backend.Search(ctx, &AssetQuery{ + startsAt := owned.StartsAt + stopsAt := owned.StopsAt + assets, err = backend.SearchOwnedAssets(ctx, &OwnedAssetQuery{ AccountId: &alicesAccountID, StartsAt: &startsAt, StopsAt: &stopsAt, - PageSize: 64, }) require.NoError(t, err) require.Len(t, assets, 1) - tooLate := owned.StopsAt.Add(time.Hour).UTC().Format(time.RFC3339) - assets, err = backend.Search(ctx, &AssetQuery{ + tooLate := owned.StopsAt + 60*60 + assets, err = backend.SearchOwnedAssets(ctx, &OwnedAssetQuery{ AccountId: &alicesAccountID, StartsAt: &startsAt, StopsAt: &tooLate, - PageSize: 64, }) require.NoError(t, err) require.Empty(t, assets, "an asset that stops too early must not be returned") // Without an account, only the assets nobody owns are listed. - assets, err = backend.Search(ctx, &AssetQuery{PageSize: 64}) + assets, err = backend.SearchListedAssets(ctx, &ListedAssetQuery{ + IA: addr.MustParseIA("1-ff00:0:110"), + StartsAt: 0, + StopsAt: math.MaxInt64, + ValidityMode: PartialOverlap, + }) require.NoError(t, err) require.Len(t, assets, 1) require.Equal(t, notOwnedAssetID, assets[0].ID) @@ -247,11 +279,12 @@ func TestCreateOrUpdateRedemptionDelegations(t *testing.T) { ctx := context.Background() backend := newTestBackend(t) ia := addr.MustParseIA("1-ff00:0:110") + currentYear := time.Now().Year() // first try creating a redemption delegation delegationToInsert := &RedemptionDelegation{ IA: ia, - Expiration: time.Date(2027, 1, 1, 0, 0, 0, 0, time.UTC), - PaidUntil: time.Date(2027, 1, 1, 0, 0, 0, 0, time.UTC), + Expiration: time.Date(currentYear+1, 1, 1, 0, 0, 0, 0, time.UTC).Unix(), + PaidUntil: time.Date(currentYear+1, 1, 1, 0, 0, 0, 0, time.UTC).Unix(), ResIdLow: 0, ResIdHigh: 1024, Key: []byte{1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16}, @@ -271,8 +304,8 @@ func TestCreateOrUpdateRedemptionDelegations(t *testing.T) { // now try to update a redemption delegation delegationToUpdate := &RedemptionDelegation{ IA: ia, - Expiration: time.Date(2028, 1, 1, 0, 0, 0, 0, time.UTC), - PaidUntil: time.Date(2028, 1, 1, 0, 0, 0, 0, time.UTC), + Expiration: time.Date(currentYear+2, 1, 1, 0, 0, 0, 0, time.UTC).Unix(), + PaidUntil: time.Date(currentYear+2, 1, 1, 0, 0, 0, 0, time.UTC).Unix(), ResIdLow: 1024, ResIdHigh: 2048, Key: []byte{16, 15, 14, 16, 13, 12, 11, 10, 9, 8, 7, 6, 5, 4, 3, 2, 1}, @@ -296,8 +329,8 @@ func TestRegisterAssetEvent(t *testing.T) { Bandwidth: 1000, BandwidthMin: 10, BandwidthMax: 1000, - StartAt: time.Date(2027, 1, 1, 0, 0, 0, 0, time.UTC), - StopsAt: time.Date(2028, 1, 1, 0, 0, 0, 0, time.UTC), + StartsAt: time.Date(2027, 1, 1, 0, 0, 0, 0, time.UTC).Unix(), + StopsAt: time.Date(2028, 1, 1, 0, 0, 0, 0, time.UTC).Unix(), TimeGranularity: 10, TimeMinDuration: 10, TimeMaxDuration: 3600, @@ -315,8 +348,8 @@ func TestRegisterAssetEvent(t *testing.T) { Bandwidth: 1000, BandwidthMin: 10, BandwidthMax: 1000, - StartAt: time.Date(2027, 1, 1, 0, 0, 0, 0, time.UTC), - StopsAt: time.Date(2027, 1, 2, 0, 0, 0, 0, time.UTC), + StartsAt: time.Date(2027, 1, 1, 0, 0, 0, 0, time.UTC).Unix(), + StopsAt: time.Date(2027, 1, 2, 0, 0, 0, 0, time.UTC).Unix(), TimeGranularity: 10, TimeMinDuration: 10, TimeMaxDuration: 3600, @@ -330,19 +363,19 @@ func TestRegisterAssetEvent(t *testing.T) { // now verify whether both registered events return the correct values. publishStats, buyStats, err := backend.SearchAssetsForStatistics(ctx, &StatisticsQuery{ IA: ia, - WindowStart: "2027-01-01T00:00:00Z", - WindowEnd: "2028-01-01T00:00:00Z", + WindowStart: time.Date(2027, 1, 1, 0, 0, 0, 0, time.UTC).Unix(), + WindowEnd: time.Date(2028, 1, 1, 0, 0, 0, 0, time.UTC).Unix(), }) require.NoError(t, err) require.Len(t, publishStats, 1) require.Len(t, buyStats, 1) require.Equal(t, publishAsset.Bandwidth, uint32(publishStats[0].Bandwidth)) require.Equal(t, publishAsset.Price, uint32(publishStats[0].Price)) - require.Equal(t, publishAsset.StartAt, publishStats[0].StartsAt) + require.Equal(t, publishAsset.StartsAt, publishStats[0].StartsAt) require.Equal(t, publishAsset.StopsAt, publishStats[0].StopsAt) require.Equal(t, buyAsset.Bandwidth, uint32(buyStats[0].Bandwidth)) require.Equal(t, buyAsset.Price, uint32(buyStats[0].Price)) - require.Equal(t, buyAsset.StartAt, buyStats[0].StartsAt) + require.Equal(t, buyAsset.StartsAt, buyStats[0].StartsAt) require.Equal(t, buyAsset.StopsAt, buyStats[0].StopsAt) } diff --git a/marketplace/db/types.go b/marketplace/db/types.go index c74eeb7268..bcea936e1f 100644 --- a/marketplace/db/types.go +++ b/marketplace/db/types.go @@ -18,7 +18,6 @@ import ( "database/sql" "encoding/binary" "math" - "time" "github.com/scionproto/scion/pkg/addr" "github.com/scionproto/scion/pkg/private/serrors" @@ -35,6 +34,13 @@ const ( AssetStateCombinePending ) +type ValidityMode int + +const ( + FullOverlap ValidityMode = 0 + PartialOverlap ValidityMode = 1 +) + type AssetEventType int const ( @@ -55,8 +61,25 @@ func (id AssetID) Int64() (int64, error) { return int64(id), nil } -// Defines the query parameters for search assets -type AssetQuery struct { +type ListedAssetQuery struct { + // ISD-AS of the asset + IA addr.IA + // ingress ID of the asset + Ingress *uint32 + // egress ID of the asset + Egress *uint32 + // the minimal required bandwidth of the asset + MinRequiredBandwidth uint32 + // starts at in unix seconds + StartsAt int64 + // stops at in unix seconds + StopsAt int64 + // the maximum price in kbps per second + Price *uint32 + ValidityMode ValidityMode +} + +type OwnedAssetQuery struct { // the account ID of the owner AccountId *int64 // ISD-AS of the asset @@ -67,16 +90,13 @@ type AssetQuery struct { Egress *uint32 // the minimal required bandwidth of the asset MinRequiredBandwidth *uint32 - // asset starts at latest in RFC3339 format - StartsAt *string - // reservation stops at earliest in RFC3339 format - StopsAt *string - // the price in kbps per second - Price *uint32 - // the page number - Page uint32 - // maximum number of returned elements - PageSize uint32 + // starts at in unix seconds + StartsAt *int64 + // stops at in unix seconds + StopsAt *int64 + // the maximum price in kbps per second + Price *uint32 + ValidityMode ValidityMode } type UsedReservationsQuery struct { @@ -92,18 +112,18 @@ type UsedReservation struct { // The reservation ID Id uint32 // inclusive start time of the reservation - StartsAt time.Time + StartsAt int64 // exclusive end time of the reservation - StopsAt time.Time + StopsAt int64 } type RedemptionDelegation struct { // ISD-AS IA addr.IA // expiration time of redemption delegation - Expiration time.Time + Expiration int64 // until when redemption delegation is paid - PaidUntil time.Time + PaidUntil int64 // the inclusive lower bound reservation ID used for redemption ResIdLow uint32 // the exclusive upper bound reservation ID used for redemption @@ -123,20 +143,20 @@ type ReservationQuery struct { Ingress *uint32 // ingress ID of the reservation Egress *uint32 - // reservation starts at latest in RFC3339 format - StartsAt *string - // reservation stops at earliest in RFC3339 format - StopsAt *string + // reservation starts at latest in unix seconds + StartsAt *int64 + // reservation stops at earliest in unix seconds + StopsAt *int64 // minimum bandwith the reservation holds Bandwidth *uint32 } type StatisticsQuery struct { // ISD-AS for which statistics should be queried IA addr.IA - // time window start in RFC3339 format - WindowStart string - // time window end in RFC3339 format - WindowEnd string + // time window start in unix seconds + WindowStart int64 + // time window end in unix seconds + WindowEnd int64 // ingress ID of the assets Ingress *uint32 // egress ID of the assets @@ -148,9 +168,9 @@ type DBStat struct { // the bandwidth of the asset Bandwidth int64 // the inclusive start time of the asset - StartsAt time.Time + StartsAt int64 // the exclusive end time of the asset - StopsAt time.Time + StopsAt int64 } type DBReservation struct { @@ -169,9 +189,9 @@ type DBReservation struct { // the dataplane encoded bandwidth of the reservation EncodedBandwidth uint16 // inclusive start time of the reservation - StartsAt time.Time + StartsAt int64 // exclusive end time of the reservation - StopsAt time.Time + StopsAt int64 // the account ID of the owner AccountId int64 // the cryptographic key @@ -192,9 +212,9 @@ type DBAsset struct { // the maximum bandwidth allowed at redemption BandwidthMax uint32 // inclusive start time of the asset - StartAt time.Time + StartsAt int64 // exclusive end time of the asset - StopsAt time.Time + StopsAt int64 // price per kbps per second of the asset Price uint32 // asset duration must be divisible by time granularity at redemption @@ -209,6 +229,12 @@ type DBAsset struct { IfIdEgress sql.NullInt32 } +type DBAssetSegment struct { + StartsAt int64 `json:"starts_at"` + StopsAt int64 `json:"stops_at"` + Available int64 `json:"available"` +} + type DBUser struct { // marketplace-wide unique user ID ID int64 diff --git a/marketplace/export_test.go b/marketplace/export_test.go index 305ec7ad78..c1cb5d0dd7 100644 --- a/marketplace/export_test.go +++ b/marketplace/export_test.go @@ -59,7 +59,8 @@ func StatisticsWindow( start, end time.Time, step, granularity time.Duration, ) (time.Time, time.Time, int) { - return statisticsWindow(start, end, step, granularity) + window_start, window_end, num_intervals := statisticsWindow(start.Unix(), end.Unix(), int64(step/time.Second), int64(granularity/time.Second)) + return time.Unix(window_start, 0).UTC(), time.Unix(window_end, 0).UTC(), int(num_intervals) } // BandwidthUtilization exposes the share of the published bandwidth that was bought. @@ -74,5 +75,5 @@ func BandwidthPerInterval( step time.Duration, numIntervals int, ) (bandwidth, income []uint64) { - return bandwidthPerInterval(assets, windowStart, windowEnd, step, numIntervals) + return bandwidthPerInterval(assets, windowStart.Unix(), windowEnd.Unix(), int64(step/time.Second), int64(numIntervals)) } diff --git a/marketplace/id_store.go b/marketplace/id_store.go index 3351514d2c..e2b6c22dcc 100644 --- a/marketplace/id_store.go +++ b/marketplace/id_store.go @@ -71,7 +71,7 @@ func (s *UsedIDStore) Init(limit_low uint32, limit_high uint32, r []*db.UsedRese s.usedIds[res.Id] = struct{}{} s.expirations = append(s.expirations, &entry{ id: res.Id, - expiration: res.StopsAt.Unix(), + expiration: res.StopsAt, }) } heap.Init(&s.expirations) diff --git a/marketplace/id_store_test.go b/marketplace/id_store_test.go index da0b717ba4..337a4b40a0 100644 --- a/marketplace/id_store_test.go +++ b/marketplace/id_store_test.go @@ -31,8 +31,8 @@ func TestIDStoreNext(t *testing.T) { err = store.Init(0, 3, []*db.UsedReservation{ { Id: 0, - StartsAt: base, - StopsAt: base.Add(time.Second * 3), + StartsAt: b, + StopsAt: b + 3, }, }) assert.NoError(t, err) diff --git a/marketplace/redeem.go b/marketplace/redeem.go index 02fbe819dd..b272f62fd4 100644 --- a/marketplace/redeem.go +++ b/marketplace/redeem.go @@ -113,7 +113,7 @@ func (h *RedemptionServerHandler) Redeem( req *hummingbird.RedeemAssetFromASRequest, ) (*hummingbird.RedeemAssetFromASResponse, error) { h.mu.Lock() - if h.local != nil && h.local.expiration.After(time.Now()) { + if h.local != nil && h.local.expiration > time.Now().Unix() { // The marketplace holds a delegation, so it redeems the asset itself. // Redeem only touches memory, so the lock is held across it. resp := h.local.Redeem(req) @@ -364,7 +364,7 @@ func (s *Service) DelegateRedemption( } handler := s.FindRedemptionServerHandler(clientID) err := handler.ApplyDelegation(ctx, &RedemptionDelegationUpdate{ - ExpirationTime: req.Msg.ExpirationTime.AsTime(), + ExpirationTime: req.Msg.ExpirationTime, IdLimitLow: req.Msg.ReservationIdLowerBound, IdLimitHigh: req.Msg.ReservationIdUpperBound, Key: req.Msg.Key, diff --git a/marketplace/redeem_test.go b/marketplace/redeem_test.go index 73809d26c7..9bf6c8564d 100644 --- a/marketplace/redeem_test.go +++ b/marketplace/redeem_test.go @@ -317,7 +317,7 @@ func TestApplyDelegationRejectsBadParams(t *testing.T) { t.Run(name, func(t *testing.T) { h := newHandler() err := h.ApplyDelegation(context.Background(), &marketplace.RedemptionDelegationUpdate{ - ExpirationTime: time.Now().Add(time.Hour), + ExpirationTime: time.Now().Add(time.Hour).Unix(), IdLimitHigh: 1000, Key: tc.key, EncodingPoints: tc.points, @@ -335,7 +335,7 @@ func TestApplyDelegationAcceptsEveryAESKeySize(t *testing.T) { for _, size := range []int{16, 24, 32} { t.Run(fmt.Sprintf("%d bytes", size), func(t *testing.T) { _, err := marketplace.NewRedemptionService(&marketplace.RedemptionDelegationUpdate{ - ExpirationTime: time.Now().Add(time.Hour), + ExpirationTime: time.Now().Add(time.Hour).Unix(), IdLimitHigh: 1000, Key: make([]byte, size), EncodingPoints: completeEncodingPoints(), @@ -349,7 +349,7 @@ func TestApplyDelegationAcceptsEveryAESKeySize(t *testing.T) { // belongs to the type, not only to the callers that happen to validate. func TestNewRedemptionServiceRejectsIncompleteEncodingPoints(t *testing.T) { _, err := marketplace.NewRedemptionService(&marketplace.RedemptionDelegationUpdate{ - ExpirationTime: time.Now().Add(time.Hour), + ExpirationTime: time.Now().Add(time.Hour).Unix(), IdLimitHigh: 1000, Key: make([]byte, 16), EncodingPoints: []uint32{100, 200}, @@ -362,7 +362,7 @@ func TestNewRedemptionServiceRejectsIncompleteEncodingPoints(t *testing.T) { // so the checks above cannot pass merely by rejecting everything. func TestNewRedemptionServiceAcceptsACompleteTable(t *testing.T) { s, err := marketplace.NewRedemptionService(&marketplace.RedemptionDelegationUpdate{ - ExpirationTime: time.Now().Add(time.Hour), + ExpirationTime: time.Now().Add(time.Hour).Unix(), IdLimitHigh: 1000, Key: make([]byte, 16), EncodingPoints: completeEncodingPoints(), diff --git a/marketplace/redemption_service.go b/marketplace/redemption_service.go index 8934337a23..475320fb4a 100644 --- a/marketplace/redemption_service.go +++ b/marketplace/redemption_service.go @@ -40,10 +40,10 @@ type RedemptionService struct { encodingPoints []uint32 cipher cipher.Block resIdStore *UsedIDStore - expiration time.Time + expiration int64 } type RedemptionDelegationUpdate struct { - ExpirationTime time.Time + ExpirationTime int64 IdLimitLow uint32 IdLimitHigh uint32 Key []byte @@ -99,7 +99,7 @@ func (r *RedemptionService) Redeem( req *hummingbird.RedeemAssetFromASRequest, ) *hummingbird.RedeemAssetFromASResponse { now := time.Now() - resId, err := r.resIdStore.Next(now.Unix(), req.StartsAt.Seconds, req.StopsAt.Seconds) + resId, err := r.resIdStore.Next(now.Unix(), req.StartsAt, req.StopsAt) if err != nil { return &hummingbird.RedeemAssetFromASResponse{ Result: &hummingbird.RedeemAssetFromASResponse_Error{ @@ -108,13 +108,13 @@ func (r *RedemptionService) Redeem( } } - unixStart := uint32(req.StartsAt.Seconds) - unixEnd := uint32(req.StopsAt.Seconds) - durSeconds := unixEnd - unixStart + unixStart := req.StartsAt + unixEnd := req.StopsAt + durSeconds := uint32(unixEnd - unixStart) encoded_bw := r.encodeBandwidth(req.Bandwidth) var buff [16]byte - ak := hbird.DeriveAuthKey(r.cipher, resId, encoded_bw, uint16(req.IngressId), uint16(req.EgressId), unixStart, uint16(durSeconds), buff[:]) + ak := hbird.DeriveAuthKey(r.cipher, resId, encoded_bw, uint16(req.IngressId), uint16(req.EgressId), uint32(unixStart), uint16(durSeconds), buff[:]) return &hummingbird.RedeemAssetFromASResponse{ Result: &hummingbird.RedeemAssetFromASResponse_ResInfo{ ResInfo: &hummingbird.ReservationInfo{ diff --git a/marketplace/service.go b/marketplace/service.go index 9e52e5787c..bc796ef15a 100644 --- a/marketplace/service.go +++ b/marketplace/service.go @@ -30,7 +30,6 @@ import ( "github.com/scionproto/scion/pkg/log" "github.com/scionproto/scion/pkg/private/serrors" "github.com/scionproto/scion/pkg/proto/hummingbird" - "google.golang.org/protobuf/types/known/timestamppb" ) type Service struct { @@ -163,11 +162,7 @@ func (s *Service) SplitAsset(ctx context.Context, req *connect.Request[hummingbi ids, err = s.store.SplitAsset(ctx, userId, assetId, bwSplit.Splits, nil) case *hummingbird.SplitAssetRequest_TimeSplit: splits := req.Msg.GetTimeSplit().Splits - times := make([]time.Time, 0, len(splits)) - for _, split := range splits { - times = append(times, split.AsTime()) - } - ids, err = s.store.SplitAsset(ctx, userId, assetId, nil, times) + ids, err = s.store.SplitAsset(ctx, userId, assetId, nil, splits) default: return nil, connect.NewError(connect.CodeInvalidArgument, serrors.New("invalid split request")) } @@ -213,16 +208,6 @@ func (s *Service) FetchReservations(ctx context.Context, req *connect.Request[hu if !ok { return nil, connect.NewError(connect.CodePermissionDenied, serrors.New("user_id not provided")) } - var startsAt *string - var stopsAt *string - if req.Msg.StartsAt != nil { - start := req.Msg.StartsAt.AsTime().UTC().Format(time.RFC3339) - startsAt = &start - } - if req.Msg.StopsAt != nil { - stop := req.Msg.StopsAt.AsTime().UTC().Format(time.RFC3339) - stopsAt = &stop - } var ia *addr.IA if req.Msg.Ia != nil { tmp := addr.IA(*req.Msg.Ia) @@ -232,8 +217,8 @@ func (s *Service) FetchReservations(ctx context.Context, req *connect.Request[hu IA: ia, Ingress: req.Msg.IngressId, Egress: req.Msg.EgressId, - StartsAt: startsAt, - StopsAt: stopsAt, + StartsAt: req.Msg.StartsAt, + StopsAt: req.Msg.StopsAt, Bandwidth: req.Msg.Bandwidth, AccountId: user, }) @@ -249,8 +234,8 @@ func (s *Service) FetchReservations(ctx context.Context, req *connect.Request[hu EgressId: reservation.Egress, Bandwidth: reservation.Bandwidth, DataplaneEncoding: uint32(reservation.EncodedBandwidth), - StartsAt: timestamppb.New(reservation.StartsAt), - StopsAt: timestamppb.New(reservation.StopsAt), + StartsAt: reservation.StartsAt, + StopsAt: reservation.StopsAt, AuthenticationKey: reservation.Key, }) } @@ -301,35 +286,10 @@ func (s *Service) UpdateAssets(ctx context.Context, req *connect.Request[humming } return 0, nil case *hummingbird.AssetUpdate_Update: - dbAsset := &db.DBAsset{ - ID: assetId, - IA: ia, - Bandwidth: t.Update.Bandwidth, - BandwidthMin: t.Update.BandwidthMin, - BandwidthMax: t.Update.BandwidthMax, - StartAt: t.Update.StartsAt.AsTime(), - StopsAt: t.Update.StopsAt.AsTime(), - Price: t.Update.Price, - TimeGranularity: t.Update.TimeGranularity, - TimeMinDuration: t.Update.TimeMinDuration, - TimeMaxDuration: t.Update.TimeMaxDuration, - } - if dbAsset.StopsAt.After(time.Now().Add(time.Duration(s.info.AssetValidityMax) * time.Second)) { + if t.Update.StopsAt != nil && time.Unix(*t.Update.StopsAt, 0).UTC().After(time.Now().Add(time.Duration(s.info.AssetValidityMax)*time.Second).UTC()) { return 0, serrors.New("validity outside allowed range") } - if t.Update.IfIdIngress != nil { - dbAsset.IfIdIngress = sql.NullInt32{ - Valid: true, - Int32: int32(*t.Update.IfIdIngress), - } - } - if t.Update.IfIdEgress != nil { - dbAsset.IfIdEgress = sql.NullInt32{ - Valid: true, - Int32: int32(*t.Update.IfIdEgress), - } - } - newId, err := s.store.UpdateListedAsset(ctx, dbAsset) + newId, err := s.store.UpdateListedAsset(ctx, assetId, ia, t.Update) if err != nil { return 0, err } @@ -340,18 +300,15 @@ func (s *Service) UpdateAssets(ctx context.Context, req *connect.Request[humming } res := make([]*hummingbird.UpdateAssetResult, 0, len(req.Msg.Assets)) for _, update := range req.Msg.Assets { - newId, err := handleAsset(update) + _, err := handleAsset(update) if err != nil { + errString := err.Error() res = append(res, &hummingbird.UpdateAssetResult{ - ResultType: &hummingbird.UpdateAssetResult_Error{ - Error: err.Error(), - }, + Error: &errString, }) } else { res = append(res, &hummingbird.UpdateAssetResult{ - ResultType: &hummingbird.UpdateAssetResult_NewId{ - NewId: protoAssetID(newId), - }, + Error: nil, }) } } @@ -372,8 +329,8 @@ func (s *Service) PublishAsset(ctx context.Context, req *connect.Request[humming Bandwidth: req.Msg.Asset.Bandwidth, BandwidthMin: req.Msg.Asset.BandwidthMin, BandwidthMax: req.Msg.Asset.BandwidthMax, - StartAt: req.Msg.Asset.StartsAt.AsTime().Truncate(time.Second), - StopsAt: req.Msg.Asset.StopsAt.AsTime().Truncate(time.Second), + StartsAt: req.Msg.Asset.StartsAt, + StopsAt: req.Msg.Asset.StopsAt, Price: req.Msg.Asset.Price, TimeGranularity: req.Msg.Asset.TimeGranularity, TimeMinDuration: req.Msg.Asset.TimeMinDuration, @@ -381,10 +338,10 @@ func (s *Service) PublishAsset(ctx context.Context, req *connect.Request[humming IfIdIngress: sql.NullInt32{}, IfIdEgress: sql.NullInt32{}, } - if dbAsset.StopsAt.After(time.Now().Add(time.Duration(s.info.AssetValidityMax) * time.Second)) { + if time.Unix(dbAsset.StopsAt, 0).UTC().After(time.Now().Add(time.Duration(s.info.AssetValidityMax) * time.Second).UTC()) { return nil, connect.NewError(connect.CodeInvalidArgument, serrors.New("validity outside allowed range")) } - if !dbAsset.StopsAt.After(dbAsset.StartAt) { + if dbAsset.StopsAt <= dbAsset.StartsAt { return nil, connect.NewError(connect.CodeInvalidArgument, serrors.New("stopsAt must come after startsAt")) } if req.Msg.Asset.IfIdIngress != nil { @@ -446,8 +403,8 @@ func (s *Service) RedeemAsset( var bw uint32 var ingressID uint32 var egressID uint32 - var startsAt time.Time - var stopsAt time.Time + var startsAt int64 + var stopsAt int64 var ia addr.IA if len(assets) == 1 { // interface pair @@ -455,30 +412,33 @@ func (s *Service) RedeemAsset( bw = pairAsset.Bandwidth ingressID = uint32(pairAsset.IfIdIngress.Int32) egressID = uint32(pairAsset.IfIdEgress.Int32) - startsAt = pairAsset.StartAt + startsAt = pairAsset.StartsAt stopsAt = pairAsset.StopsAt ia = pairAsset.IA } else if len(assets) == 2 { // ingress and egress assets ingressAsset := assets[0] egressAsset := assets[1] - bw = min(ingressAsset.Bandwidth, egressAsset.Bandwidth) + if ingressAsset.IA != egressAsset.IA || + ingressAsset.Bandwidth != egressAsset.Bandwidth || + ingressAsset.StartsAt != egressAsset.StartsAt || + ingressAsset.StopsAt != egressAsset.StopsAt || + !ingressAsset.IfIdIngress.Valid || + !egressAsset.IfIdEgress.Valid { + return nil, connect.NewError(connect.CodeInvalidArgument, serrors.New("invalid assets")) + } + bw = ingressAsset.Bandwidth ingressID = uint32(ingressAsset.IfIdIngress.Int32) egressID = uint32(egressAsset.IfIdEgress.Int32) ia = ingressAsset.IA - if ingressAsset.StartAt.Before(egressAsset.StartAt) { - startsAt = egressAsset.StartAt - } else { - startsAt = ingressAsset.StartAt - } - if ingressAsset.StopsAt.Before(egressAsset.StopsAt) { - stopsAt = ingressAsset.StopsAt - } else { - stopsAt = egressAsset.StopsAt - } + startsAt = ingressAsset.StartsAt + stopsAt = ingressAsset.StopsAt } else { return nil, connect.NewError(connect.CodeInvalidArgument, serrors.New("invalid assets")) } + if stopsAt <= startsAt { + return nil, connect.NewError(connect.CodeInvalidArgument, serrors.New("invalid validity range")) + } undoRedemption := func() error { // we use context.Background here because if the client disconnected in the meantime, // we cannot undo the redemption using the request's context. @@ -496,10 +456,6 @@ func (s *Service) RedeemAsset( } return err } - if stopsAt.Before(startsAt) { - return nil, connect.NewError(connect.CodeInvalidArgument, - serrors.Join(serrors.New("stops at before starts at"), undoRedemption())) - } peer := s.FindRedemptionServerHandler(ia) // The deadline is derived from the request context: a client who disconnects releases // the asset instead of holding it until the timeout. @@ -509,8 +465,8 @@ func (s *Service) RedeemAsset( Bandwidth: bw, IngressId: ingressID, EgressId: egressID, - StartsAt: timestamppb.New(startsAt), - StopsAt: timestamppb.New(stopsAt), + StartsAt: startsAt, + StopsAt: stopsAt, }) if err != nil { return nil, connect.NewError(connect.CodeUnavailable, @@ -562,19 +518,19 @@ func (s *Service) RedeemAsset( // is preserved. Without that, windowEnd.Sub(windowStart) is negative and the interval // count with it, which makes the slices of the caller panic. func statisticsWindow( - start, end time.Time, - step, granularity time.Duration, -) (windowStart, windowEnd time.Time, numIntervals int) { - if end.Before(start) { + start, end int64, + step, granularity int64, +) (windowStart, windowEnd int64, numIntervals int64) { + if end < start { start, end = end, start } - windowStart = start.UTC().Truncate(granularity) - windowEnd = hbird.RoundUpTime(end.UTC(), granularity) + windowStart = start - (start % granularity) + windowEnd = hbird.RoundUpTime(time.Unix(end, 0).UTC(), time.Duration(granularity)*time.Second).Unix() if step <= 0 { // Only reachable with a granularity of 0, which the config does not allow. return windowStart, windowEnd, 0 } - return windowStart, windowEnd, int(windowEnd.Sub(windowStart) / step) + return windowStart, windowEnd, (windowEnd - windowStart) / step } // bandwidthUtilization is the share of the published bandwidth that was bought. @@ -594,41 +550,41 @@ func bandwidthUtilization(bought, published uint64) float64 { // window it overlaps, in kbps seconds, along with the revenue that bandwidth stands for. func bandwidthPerInterval( assets []*db.DBStat, - windowStart, windowEnd time.Time, - step time.Duration, - numIntervals int, + windowStart, windowEnd int64, + step int64, + numIntervals int64, ) (bandwidth, income []uint64) { bandwidth = make([]uint64, numIntervals) income = make([]uint64, numIntervals) for _, asset := range assets { start := asset.StartsAt stop := asset.StopsAt - if asset.StartsAt.Before(windowStart) { + if asset.StartsAt < windowStart { start = windowStart } - if asset.StopsAt.After(windowEnd) { + if asset.StopsAt > windowEnd { stop = windowEnd } - first := int(start.Sub(windowStart) / step) - last := int(stop.Sub(windowStart) / step) + first := int64((start - windowStart) / step) + last := int64((stop - windowStart) / step) if first < 0 || last > numIntervals { continue } - if stop.Equal(windowStart.Add(time.Duration(last) * step)) { + if stop == (windowStart + (last * step)) { last-- } for i := first; i <= last && i < numIntervals; i++ { - intervalStart := windowStart.Add(time.Duration(i) * step) - intervalEnd := intervalStart.Add(step) + intervalStart := windowStart + (i * step) + intervalEnd := intervalStart + step overlapStart := start overlapEnd := stop - if asset.StartsAt.Before(intervalStart) { + if asset.StartsAt < intervalStart { overlapStart = intervalStart } - if asset.StopsAt.After(intervalEnd) { + if asset.StopsAt > intervalEnd { overlapEnd = intervalEnd } - duration := uint64(overlapEnd.Sub(overlapStart).Seconds()) + duration := uint64(overlapEnd - overlapStart) bwTimesDuration := uint64(asset.Bandwidth) * duration bandwidth[i] += bwTimesDuration income[i] += bwTimesDuration * uint64(asset.Price) @@ -648,15 +604,15 @@ func (s *Service) Statistics( granularity := time.Duration(s.info.StatisticsTimeGranularity) * time.Second step := hbird.RoundUpDuration(time.Duration(req.Msg.Step)*time.Second, granularity) windowStart, windowEnd, numIntervals := statisticsWindow( - req.Msg.Start.AsTime(), req.Msg.End.AsTime(), step, granularity) + req.Msg.Start, req.Msg.End, int64(step), int64(granularity)) if numIntervals > 1024 { return nil, connect.NewError(connect.CodeResourceExhausted, serrors.New("too many intervals")) } publishedAssets, boughtAssets, err := s.store.Statistics(ctx, &db.StatisticsQuery{ IA: ia, - WindowStart: windowStart.Format(time.RFC3339), - WindowEnd: windowEnd.Format(time.RFC3339), + WindowStart: windowStart, + WindowEnd: windowEnd, Ingress: req.Msg.IfIdIngress, Egress: req.Msg.IfIdEgress, }) @@ -664,9 +620,9 @@ func (s *Service) Statistics( return nil, connect.NewError(connect.CodeInvalidArgument, err) } bwPublished, _ := bandwidthPerInterval( - publishedAssets, windowStart, windowEnd, step, numIntervals) + publishedAssets, windowStart, windowEnd, int64(step/time.Second), numIntervals) bwBought, income := bandwidthPerInterval( - boughtAssets, windowStart, windowEnd, step, numIntervals) + boughtAssets, windowStart, windowEnd, int64(step/time.Second), numIntervals) respEntries := make([]*hummingbird.StatisticsResponseEntry, numIntervals) for i := range numIntervals { @@ -686,75 +642,82 @@ func (s *Service) SearchAssets(ctx context.Context, req *connect.Request[humming var owner_id *int64 switch user := ctx.Value("user").(type) { case int64: - if req.Msg.Owned { - owner_id = &user - } + owner_id = &user case addr.IA: default: return nil, connect.NewError(connect.CodePermissionDenied, serrors.New("user_id not provided")) } - var startsAt *string - var stopsAt *string - if req.Msg.StartsAtLatest != nil { - start := req.Msg.StartsAtLatest.AsTime().UTC().Format(time.RFC3339) - startsAt = &start - } - if req.Msg.StopsAtEarliest != nil { - stop := req.Msg.StopsAtEarliest.AsTime().UTC().Format(time.RFC3339) - stopsAt = &stop - } - var ia *addr.IA - if req.Msg.Ia != nil { - tmp := addr.IA(*req.Msg.Ia) - ia = &tmp - } - pageSize := s.info.MaxReturnedAssets - if pageSize == 0 { - return nil, connect.NewError(connect.CodeInvalidArgument, serrors.New("page size cannot be 0")) - } - if req.Msg.MaxReturnedAssets != nil { - pageSize = min(pageSize, *req.Msg.MaxReturnedAssets) + var assets []*db.DBAsset + var err error + switch t := req.Msg.Request.(type) { + case *hummingbird.SearchAssetsRequest_Listed: + assets, err = s.store.SearchListedAssets(ctx, &db.ListedAssetQuery{ + IA: addr.IA(t.Listed.Ia), + Ingress: t.Listed.IfIdIngress, + Egress: t.Listed.IfIdEgress, + MinRequiredBandwidth: t.Listed.MinRequiredBw, + Price: t.Listed.Price, + StartsAt: t.Listed.StartsAtLatest, + StopsAt: t.Listed.StopsAtEarliest, + ValidityMode: db.ValidityMode(req.Msg.ValidityMode), + }) + case *hummingbird.SearchAssetsRequest_Owned: + assets, err = s.store.SearchOwnedAssets(ctx, &db.OwnedAssetQuery{ + AccountId: owner_id, + IA: (*addr.IA)(t.Owned.Ia), + Ingress: t.Owned.IfIdIngress, + Egress: t.Owned.IfIdEgress, + MinRequiredBandwidth: t.Owned.MinRequiredBw, + Price: t.Owned.Price, + StartsAt: t.Owned.StartsAtLatest, + StopsAt: t.Owned.StopsAtEarliest, + ValidityMode: db.ValidityMode(req.Msg.ValidityMode), + }) + default: + return nil, connect.NewError(connect.CodeInvalidArgument, serrors.New("invalid request")) } - - assets, err := s.store.Search(ctx, &db.AssetQuery{ - AccountId: owner_id, - IA: ia, - Ingress: req.Msg.IfIdIngress, - Egress: req.Msg.IfIdEgress, - MinRequiredBandwidth: req.Msg.MinRequiredBw, - Price: req.Msg.Price, - StartsAt: startsAt, - StopsAt: stopsAt, - Page: req.Msg.Page, - PageSize: pageSize, - }) if err != nil { return nil, connect.NewError(connect.CodeInvalidArgument, err) } repAssets := make([]*hummingbird.SearchAsset, 0, len(assets)) + assetMap := make(map[int64]*hummingbird.SearchAsset) for _, asset := range assets { - a := &hummingbird.SearchAsset{ - AssetId: protoAssetID(asset.ID), - Ia: uint64(asset.IA), - Bandwidth: asset.Bandwidth, - BandwidthMin: asset.BandwidthMin, - BandwidthMax: asset.BandwidthMax, - TimeMinDuration: asset.TimeMinDuration, - TimeMaxDuration: asset.TimeMaxDuration, - StartsAt: timestamppb.New(asset.StartAt), - StopsAt: timestamppb.New(asset.StopsAt), - TimeGranularity: asset.TimeGranularity, - Price: asset.Price, - } - if asset.IfIdIngress.Valid { - ingress := uint32(asset.IfIdIngress.Int32) - a.IfIdIngress = &ingress - } - if asset.IfIdEgress.Valid { - egress := uint32(asset.IfIdEgress.Int32) - a.IfIdEgress = &egress + mapAsset, found := assetMap[asset.ID] + if found { + mapAsset.Intervals = append(mapAsset.Intervals, &hummingbird.SearchAssetInterval{ + Bandwidth: asset.Bandwidth, + StartsAt: asset.StartsAt, + StopsAt: asset.StopsAt, + }) + } else { + a := &hummingbird.SearchAsset{ + AssetId: protoAssetID(asset.ID), + Ia: uint64(asset.IA), + BandwidthMin: asset.BandwidthMin, + BandwidthMax: asset.BandwidthMax, + TimeMinDuration: asset.TimeMinDuration, + TimeMaxDuration: asset.TimeMaxDuration, + TimeGranularity: asset.TimeGranularity, + Price: asset.Price, + Intervals: []*hummingbird.SearchAssetInterval{ + &hummingbird.SearchAssetInterval{ + Bandwidth: asset.Bandwidth, + StartsAt: asset.StartsAt, + StopsAt: asset.StopsAt, + }, + }, + } + if asset.IfIdIngress.Valid { + ingress := uint32(asset.IfIdIngress.Int32) + a.IfIdIngress = &ingress + } + if asset.IfIdEgress.Valid { + egress := uint32(asset.IfIdEgress.Int32) + a.IfIdEgress = &egress + } + repAssets = append(repAssets, a) + assetMap[asset.ID] = a } - repAssets = append(repAssets, a) } return &connect.Response[hummingbird.SearchAssetsResponse]{ Msg: &hummingbird.SearchAssetsResponse{ diff --git a/marketplace/statistics_test.go b/marketplace/statistics_test.go index 6427b87085..c0b251896a 100644 --- a/marketplace/statistics_test.go +++ b/marketplace/statistics_test.go @@ -126,10 +126,10 @@ func TestStatisticsWindow(t *testing.T) { func TestBandwidthPerIntervalGaps(t *testing.T) { // Two one-day assets, days 0 and 4 of a six day window. assets := []*db.DBStat{ - {Bandwidth: 100, Price: 2, StartsAt: statsBase, StopsAt: statsBase.Add(day)}, + {Bandwidth: 100, Price: 2, StartsAt: statsBase.Unix(), StopsAt: statsBase.Add(day).Unix()}, { Bandwidth: 100, Price: 2, - StartsAt: statsBase.Add(4 * day), StopsAt: statsBase.Add(5 * day), + StartsAt: statsBase.Add(4 * day).Unix(), StopsAt: statsBase.Add(5 * day).Unix(), }, } windowStart, windowEnd, n := marketplace.StatisticsWindow( @@ -171,7 +171,7 @@ func TestBandwidthPerIntervalEmptyWindow(t *testing.T) { // so that the extraction of the loop is covered for both of its outputs. func TestBandwidthPerIntervalRevenue(t *testing.T) { assets := []*db.DBStat{ - {Bandwidth: 100, Price: 3, StartsAt: statsBase, StopsAt: statsBase.Add(day)}, + {Bandwidth: 100, Price: 3, StartsAt: statsBase.Unix(), StopsAt: statsBase.Add(day).Unix()}, } windowStart, windowEnd, n := marketplace.StatisticsWindow( statsBase, statsBase.Add(day), day, day) diff --git a/marketplace/storage/BUILD.bazel b/marketplace/storage/BUILD.bazel index 6e29c21fdb..48986fb9c7 100644 --- a/marketplace/storage/BUILD.bazel +++ b/marketplace/storage/BUILD.bazel @@ -13,7 +13,6 @@ go_library( "//marketplace/db:go_default_library", "//pkg/addr:go_default_library", "//pkg/hummingbird:go_default_library", - "//pkg/hummingbird/marketplace:go_default_library", "//pkg/private/serrors:go_default_library", "//pkg/proto/hummingbird:go_default_library", "//private/config:go_default_library", diff --git a/marketplace/storage/split.go b/marketplace/storage/split.go index f12bc7bb81..617d7ca55e 100644 --- a/marketplace/storage/split.go +++ b/marketplace/storage/split.go @@ -16,18 +16,17 @@ package storage import ( "fmt" - "time" ) type RequestedSplit struct { - ExactFrom time.Time - ExactTo time.Time + ExactFrom int64 + ExactTo int64 ExactBandwidth uint32 } type AssetSegment struct { - StartsAt time.Time - StopsAt time.Time + StartsAt int64 + StopsAt int64 Bandwidth uint32 } @@ -37,30 +36,24 @@ type SplitResult struct { } func validateSplit(asset AssetSegment, p RequestedSplit) error { - if p.ExactFrom.Before(asset.StartsAt) || p.ExactTo.After(asset.StopsAt) { + if p.ExactFrom < asset.StartsAt || p.ExactTo > asset.StopsAt { return fmt.Errorf("split outside asset bounds") } - if !p.ExactFrom.Before(p.ExactTo) { + if p.ExactFrom >= p.ExactTo { return fmt.Errorf("invalid validity range") } - if p.ExactFrom.Nanosecond() != 0 || p.ExactTo.Nanosecond() != 0 { - return fmt.Errorf("timestamps must be second precision") - } if p.ExactBandwidth > asset.Bandwidth { return fmt.Errorf("split amount exceeds asset amount") } if p.ExactBandwidth == 0 { return fmt.Errorf("split cannot be zero") } - if p.ExactFrom.Equal(p.ExactTo) { - return fmt.Errorf("split cannot be zero") - } return nil } func SplitAsset(asset AssetSegment, split RequestedSplit) (*SplitResult, error) { - if !asset.StartsAt.Before(asset.StopsAt) { + if asset.StartsAt >= asset.StopsAt { return nil, fmt.Errorf("invalid asset range") } if err := validateSplit(asset, split); err != nil { @@ -80,9 +73,9 @@ func SplitAsset(asset AssetSegment, split RequestedSplit) (*SplitResult, error) }) remainingAsset.Bandwidth = split.ExactBandwidth } - if split.ExactFrom.Equal(remainingAsset.StartsAt) { + if split.ExactFrom == remainingAsset.StartsAt { // no left remainder asset exists - if split.ExactTo.Equal(remainingAsset.StopsAt) { + if split.ExactTo == remainingAsset.StopsAt { // no right remainder asset exists -> split = asset splitResult.Split = AssetSegment{ Bandwidth: remainingAsset.Bandwidth, @@ -104,7 +97,7 @@ func SplitAsset(asset AssetSegment, split RequestedSplit) (*SplitResult, error) } } else { // a left remainder asset exists - if split.ExactTo.Equal(remainingAsset.StopsAt) { + if split.ExactTo == remainingAsset.StopsAt { // no right remainder asset exists -> only a left remainder exists splitResult.Split = AssetSegment{ Bandwidth: remainingAsset.Bandwidth, diff --git a/marketplace/storage/split_test.go b/marketplace/storage/split_test.go index dcacd22beb..d8b7b72a1f 100644 --- a/marketplace/storage/split_test.go +++ b/marketplace/storage/split_test.go @@ -26,8 +26,8 @@ import ( // and what goes back on the market. The remainders must never overlap the // purchase, otherwise the same capacity could be sold twice. func TestSplitAsset(t *testing.T) { - start := time.Date(2026, time.January, 1, 0, 0, 0, 0, time.UTC) - stop := start.Add(time.Hour) + start := time.Date(2026, time.January, 1, 0, 0, 0, 0, time.UTC).Unix() + stop := start + 60*60 asset := func() AssetSegment { return AssetSegment{ Bandwidth: 1000, @@ -47,48 +47,48 @@ func TestSplitAsset(t *testing.T) { }, "tail, only a left remainder": { split: RequestedSplit{ - ExactFrom: start.Add(10 * time.Minute), + ExactFrom: start + (10 * 60), ExactTo: stop, ExactBandwidth: 1000, }, expected: AssetSegment{ - StartsAt: start.Add(10 * time.Minute), + StartsAt: start + (10 * 60), StopsAt: stop, Bandwidth: 1000, }, remainders: []AssetSegment{ - {StartsAt: start, StopsAt: start.Add(10 * time.Minute), Bandwidth: 1000}, + {StartsAt: start, StopsAt: start + (10 * 60), Bandwidth: 1000}, }, }, "head, only a right remainder": { split: RequestedSplit{ ExactFrom: start, - ExactTo: start.Add(10 * time.Minute), + ExactTo: start + (10 * 60), ExactBandwidth: 1000, }, expected: AssetSegment{ StartsAt: start, - StopsAt: start.Add(10 * time.Minute), + StopsAt: start + (10 * 60), Bandwidth: 1000, }, remainders: []AssetSegment{ - {StartsAt: start.Add(10 * time.Minute), StopsAt: stop, Bandwidth: 1000}, + {StartsAt: start + (10 * 60), StopsAt: stop, Bandwidth: 1000}, }, }, "middle, a left and a right remainder": { split: RequestedSplit{ - ExactFrom: start.Add(10 * time.Minute), - ExactTo: start.Add(20 * time.Minute), + ExactFrom: start + (10 * 60), + ExactTo: start + (20 * 60), ExactBandwidth: 1000, }, expected: AssetSegment{ - StartsAt: start.Add(10 * time.Minute), - StopsAt: start.Add(20 * time.Minute), + StartsAt: start + (10 * 60), + StopsAt: start + (20 * 60), Bandwidth: 1000, }, remainders: []AssetSegment{ - {StartsAt: start, StopsAt: start.Add(10 * time.Minute), Bandwidth: 1000}, - {StartsAt: start.Add(20 * time.Minute), StopsAt: stop, Bandwidth: 1000}, + {StartsAt: start, StopsAt: start + (10 * 60), Bandwidth: 1000}, + {StartsAt: start + (20 * 60), StopsAt: stop, Bandwidth: 1000}, }, }, "part of the bandwidth of the whole asset": { @@ -100,20 +100,20 @@ func TestSplitAsset(t *testing.T) { }, "part of the bandwidth of a middle slice": { split: RequestedSplit{ - ExactFrom: start.Add(10 * time.Minute), - ExactTo: start.Add(20 * time.Minute), + ExactFrom: start + (10 * 60), + ExactTo: start + (20 * 60), ExactBandwidth: 400, }, expected: AssetSegment{ - StartsAt: start.Add(10 * time.Minute), - StopsAt: start.Add(20 * time.Minute), + StartsAt: start + (10 * 60), + StopsAt: start + (20 * 60), Bandwidth: 400, }, remainders: []AssetSegment{ // The bandwidth that was not bought, for the whole window. {StartsAt: start, StopsAt: stop, Bandwidth: 600}, - {StartsAt: start, StopsAt: start.Add(10 * time.Minute), Bandwidth: 400}, - {StartsAt: start.Add(20 * time.Minute), StopsAt: stop, Bandwidth: 400}, + {StartsAt: start, StopsAt: start + (10 * 60), Bandwidth: 400}, + {StartsAt: start + (20 * 60), StopsAt: stop, Bandwidth: 400}, }, }, } @@ -128,8 +128,8 @@ func TestSplitAsset(t *testing.T) { // What is sold and what is left must add up to the asset, so no // remainder may overlap the purchase in both time and bandwidth. for _, remainder := range result.Remainders { - overlapsInTime := remainder.StartsAt.Before(result.Split.StopsAt) && - remainder.StopsAt.After(result.Split.StartsAt) + overlapsInTime := remainder.StartsAt < (result.Split.StopsAt) && + remainder.StopsAt > (result.Split.StartsAt) if !overlapsInTime { continue } @@ -144,16 +144,16 @@ func TestSplitAsset(t *testing.T) { // TestSplitAssetRejects checks that SplitAsset returns error when the asset cannot be split into // the required parts. func TestSplitAssetRejects(t *testing.T) { - start := time.Date(2026, time.January, 1, 0, 0, 0, 0, time.UTC) - stop := start.Add(time.Hour) + start := time.Date(2026, time.January, 1, 0, 0, 0, 0, time.UTC).Unix() + stop := start + 60*60 asset := AssetSegment{Bandwidth: 1000, StartsAt: start, StopsAt: stop} testCases := map[string]RequestedSplit{ "starts before the asset": { - ExactFrom: start.Add(-time.Second), ExactTo: stop, ExactBandwidth: 1000, + ExactFrom: start - 1, ExactTo: stop, ExactBandwidth: 1000, }, "stops after the asset": { - ExactFrom: start, ExactTo: stop.Add(time.Second), ExactBandwidth: 1000, + ExactFrom: start, ExactTo: stop + 1, ExactBandwidth: 1000, }, "empty range": { ExactFrom: start, ExactTo: start, ExactBandwidth: 1000, @@ -161,9 +161,6 @@ func TestSplitAssetRejects(t *testing.T) { "reversed range": { ExactFrom: stop, ExactTo: start, ExactBandwidth: 1000, }, - "sub second precision": { - ExactFrom: start.Add(time.Millisecond), ExactTo: stop, ExactBandwidth: 1000, - }, "more bandwidth than the asset has": { ExactFrom: start, ExactTo: stop, ExactBandwidth: 1001, }, diff --git a/marketplace/storage/storage.go b/marketplace/storage/storage.go index a5ac35939c..b581f4c88b 100644 --- a/marketplace/storage/storage.go +++ b/marketplace/storage/storage.go @@ -24,7 +24,6 @@ import ( marketplacedb "github.com/scionproto/scion/marketplace/db" "github.com/scionproto/scion/pkg/addr" hbird "github.com/scionproto/scion/pkg/hummingbird" - hbirdmarketplace "github.com/scionproto/scion/pkg/hummingbird/marketplace" "github.com/scionproto/scion/pkg/private/serrors" "github.com/scionproto/scion/pkg/proto/hummingbird" "github.com/scionproto/scion/private/config" @@ -70,11 +69,18 @@ func NewStorage( }, nil } -func (s *MarketplaceStorage) Search( +func (s *MarketplaceStorage) SearchListedAssets( ctx context.Context, - params *marketplacedb.AssetQuery, + params *marketplacedb.ListedAssetQuery, ) ([]*marketplacedb.DBAsset, error) { - return s.db.Search(ctx, params) + return s.db.SearchListedAssets(ctx, params) +} + +func (s *MarketplaceStorage) SearchOwnedAssets( + ctx context.Context, + params *marketplacedb.OwnedAssetQuery, +) ([]*marketplacedb.DBAsset, error) { + return s.db.SearchOwnedAssets(ctx, params) } func (s *MarketplaceStorage) FetchReservations( @@ -98,7 +104,11 @@ func (s *MarketplaceStorage) PublishAsset( if err != nil { return err } - _, err := tx.RegisterAssetEvent(ctx, a, marketplacedb.AssetPublished) + _, err = tx.InsertAssetSegment(ctx, assetId, a.StartsAt, a.StopsAt, a.Bandwidth) + if err != nil { + return err + } + _, err = tx.RegisterAssetEvent(ctx, a, marketplacedb.AssetPublished) if err != nil { return err } @@ -108,22 +118,74 @@ func (s *MarketplaceStorage) PublishAsset( } func (s *MarketplaceStorage) UpdateListedAsset( ctx context.Context, - a *marketplacedb.DBAsset, + id int64, + ia addr.IA, + a *hummingbird.UpdateAsset, ) (int64, error) { - if err := validateAsset(a); err != nil { - return 0, err - } var newID int64 err := s.db.WithTx(ctx, func(tx marketplacedb.Repository) error { - x, err := tx.DeleteListedAsset(ctx, a.IA, a.ID) + // get current dbAsset + dbAsset, err := tx.LookupAsset(ctx, id, ia) + if err != nil { + return err + } + // check for illegal modifications + if (a.Bandwidth != nil && *a.Bandwidth < dbAsset.Bandwidth) || + (a.StartsAt != nil && *a.StartsAt > dbAsset.StartsAt) || + (a.StopsAt != nil && *a.StopsAt < dbAsset.StopsAt) || + (a.TimeMinDuration != nil && *a.TimeMinDuration > dbAsset.TimeMinDuration) || + (a.TimeMaxDuration != nil && *a.TimeMaxDuration < dbAsset.TimeMaxDuration) || + (a.BandwidthMin != nil && *a.BandwidthMin > dbAsset.BandwidthMin) || + (a.BandwidthMax != nil && *a.BandwidthMax < dbAsset.BandwidthMax) { + return serrors.New("invalid asset modifications") + } + if a.Bandwidth != nil && *a.Bandwidth > dbAsset.Bandwidth { + if err = tx.IncreaseAssetSegmentsBandwidth(ctx, id, int64(*a.Bandwidth-dbAsset.Bandwidth)); err != nil { + return err + } + dbAsset.Bandwidth = *a.Bandwidth + } + if a.StartsAt != nil && *a.StartsAt < dbAsset.StartsAt { + if _, err = tx.InsertAssetSegment(ctx, id, *a.StartsAt, dbAsset.StartsAt, dbAsset.Bandwidth); err != nil { + return err + } + if err = tx.MergeAssetAt(ctx, id, dbAsset.StartsAt); err != nil { + return err + } + dbAsset.StartsAt = *a.StartsAt + } + if a.StopsAt != nil && *a.StopsAt > dbAsset.StopsAt { + if _, err = tx.InsertAssetSegment(ctx, id, dbAsset.StopsAt, *a.StopsAt, dbAsset.Bandwidth); err != nil { + return err + } + if err = tx.MergeAssetAt(ctx, id, dbAsset.StopsAt); err != nil { + return err + } + dbAsset.StopsAt = *a.StopsAt + } + if a.BandwidthMin != nil { + dbAsset.BandwidthMin = *a.BandwidthMin + } + if a.BandwidthMax != nil { + dbAsset.BandwidthMax = *a.BandwidthMax + } + if a.TimeMinDuration != nil { + dbAsset.TimeMinDuration = *a.TimeMinDuration + } + if a.TimeMaxDuration != nil { + dbAsset.TimeMaxDuration = *a.TimeMaxDuration + } + if a.Price != nil { + dbAsset.Price = *a.Price + } + x, err := tx.UpdateAsset(ctx, dbAsset) if err != nil { return err } if x != 1 { - return serrors.New("no modifiable asset with that ID found") + return serrors.New("invalid assets") } - newID, err = tx.InsertAsset(ctx, a) - return err + return nil }) return newID, err } @@ -295,24 +357,24 @@ func (s *MarketplaceStorage) CreateOrUpdateRedemptionDelegations( if err != nil { return err } - paidUntil := time.Now() + paidUntil := time.Now().Unix() if delegation != nil { paidUntil = delegation.PaidUntil } if s.delegationHourlyFee == 0 { - if r.Expiration.After(paidUntil) { + if r.Expiration > paidUntil { r.PaidUntil = r.Expiration } else { r.PaidUntil = paidUntil } - } else if r.Expiration.After(paidUntil) { - paymentDuration := hbird.RoundUpDuration(r.Expiration.Sub(paidUntil), time.Hour) + } else if r.Expiration > paidUntil { + paymentDuration := hbird.RoundUpDuration(time.Duration(r.Expiration-paidUntil)*time.Second, time.Hour) numHours := uint64(paymentDuration / time.Hour) if _, err := tx.UpdateASMoney( ctx, r.IA, -int64(numHours*s.delegationHourlyFee)); err != nil { return err } - r.PaidUntil = paidUntil.Add(paymentDuration) + r.PaidUntil = paidUntil + int64(numHours)*60*60 } else { r.PaidUntil = paidUntil } @@ -338,6 +400,33 @@ func addInt64( return a + b, true } +func multiplyInts(xs ...int64) (int64, bool) { + result := int64(1) + + for _, x := range xs { + if x == 0 { + return 0, true + } + if result > 0 { + if x > 0 && result > math.MaxInt64/x { + return 0, false + } + if x < 0 && x < math.MinInt64/result { + return 0, false + } + } else { + if x > 0 && result < math.MinInt64/x { + return 0, false + } + if x < 0 && result < math.MaxInt64/x { + return 0, false + } + } + result *= x + } + return result, true +} + func (s *MarketplaceStorage) IncrementASJWTVersion( ctx context.Context, ia addr.IA, @@ -410,7 +499,7 @@ func (s *MarketplaceStorage) CombineAssets( ingress := assets[0].IfIdIngress egress := assets[0].IfIdEgress combinedStop := assets[0].StopsAt - useTimeAxis := assets[0].StopsAt.Equal(assets[1].StartAt) + useTimeAxis := assets[0].StopsAt == assets[1].StartsAt combinedMinBW := assets[0].BandwidthMin combinedMaxBW := assets[0].BandwidthMax combinedTimeMinDuration := assets[0].TimeMinDuration @@ -437,12 +526,12 @@ func (s *MarketplaceStorage) CombineAssets( if assets[0].Bandwidth != asset.Bandwidth { return serrors.New("assets cannot be combined") } - if !combinedStop.Equal(asset.StartAt) { + if combinedStop != asset.StartsAt { return serrors.New("assets cannot be combined") } combinedStop = asset.StopsAt } else { - if !asset.StartAt.Equal(assets[0].StartAt) || !asset.StopsAt.Equal(assets[0].StopsAt) { + if asset.StartsAt != assets[0].StartsAt || asset.StopsAt != assets[0].StopsAt { return serrors.New("assets cannot be combined") } combinedBandwidth += asset.Bandwidth @@ -462,7 +551,7 @@ func (s *MarketplaceStorage) CombineAssets( TimeMaxDuration: combinedTimeMaxDuration, TimeGranularity: combinedTimeGranularity, Bandwidth: combinedBandwidth, - StartAt: assets[0].StartAt, + StartsAt: assets[0].StartsAt, StopsAt: combinedStop, } for _, assetId := range assetIds { @@ -481,7 +570,7 @@ func (s *MarketplaceStorage) SplitAsset( accountID int64, assetId int64, bwSplit []uint32, - timeSplit []time.Time, + timeSplit []int64, ) ([]int64, error) { if len(bwSplit) == 0 && len(timeSplit) == 0 { return nil, serrors.New("invalid split request") @@ -499,7 +588,7 @@ func (s *MarketplaceStorage) SplitAsset( } var splitSegments []AssetSegment baseSegment := AssetSegment{ - StartsAt: dbAsset.StartAt, + StartsAt: dbAsset.StartsAt, StopsAt: dbAsset.StopsAt, Bandwidth: dbAsset.Bandwidth, } @@ -542,7 +631,7 @@ func (s *MarketplaceStorage) SplitAsset( } } } - if baseSegment.Bandwidth != 0 && !baseSegment.StartsAt.Equal(baseSegment.StopsAt) { + if baseSegment.Bandwidth != 0 && baseSegment.StartsAt != baseSegment.StopsAt { splitSegments = append(splitSegments, baseSegment) } if err := tx.RemoveAsset(ctx, assetId); err != nil { @@ -561,7 +650,7 @@ func (s *MarketplaceStorage) SplitAsset( IfIdIngress: dbAsset.IfIdIngress, IfIdEgress: dbAsset.IfIdEgress, Bandwidth: segment.Bandwidth, - StartAt: segment.StartsAt, + StartsAt: segment.StartsAt, StopsAt: segment.StopsAt, } splitAssetId, err := tx.InsertAsset(ctx, asset) @@ -603,130 +692,98 @@ func (s *MarketplaceStorage) BuyAssets( assets []*hummingbird.BuyAsset, maxPrice uint64, ) ([]int64, int64, error) { - uniqueCheck := make(map[int64]bool) + if len(assets) == 0 { + return nil, 0, serrors.New("no assets specified") + } for _, asset := range assets { - assetId, err := DatabaseAssetID(asset.AssetId) - if err != nil { - return nil, 0, err - } - if uniqueCheck[assetId] { - return nil, 0, serrors.New("Only a single split per asset per buy request allowed") - } - if asset.StopsAtExactly.AsTime().Before(asset.StartsAtExactly.AsTime()) { - return nil, 0, serrors.New("End of validity must come after start of validity") + if asset.BandwidthExact <= 0 || asset.StartsAtExactly >= asset.StopsAtExactly { + return nil, 0, serrors.New("invalid asset") } - uniqueCheck[assetId] = true } - boughtAssets := make([]int64, 0, 1) + boughtIDs := []int64{} costAcc := int64(0) + err := s.db.WithTx(ctx, func(tx marketplacedb.Repository) error { - for _, asset := range assets { - assetId, err := DatabaseAssetID(asset.AssetId) + calcCost := func(a *marketplacedb.DBAsset) (int64, bool) { + totalAssetPrice, ok := multiplyInts(int64(a.StopsAt-a.StartsAt), + int64(a.Bandwidth), int64(a.Price)) + if !ok { + return 0, false + } + fee := int64(float64(totalAssetPrice)*float64(s.transactionFeeRelative)) + + int64(s.transactionFeeAbsolute) + return addInt64(int64(totalAssetPrice), fee) + } + buyAsset := func( + assetId int64, + startsAt int64, + stopsAt int64, + bw uint32, + ) (*marketplacedb.DBAsset, int64, error) { + asset, err := tx.CheckAssetAvailability(ctx, assetId, startsAt, stopsAt, bw) if err != nil { - return err + return nil, 0, err + } + if err = tx.SplitAssetAt(ctx, assetId, startsAt); err != nil { + return nil, 0, err + } + if err = tx.SplitAssetAt(ctx, assetId, stopsAt); err != nil { + return nil, 0, err } - dbAsset, err := tx.TransitionAsset(ctx, assetId, nil, - marketplacedb.AssetStateAvailable, marketplacedb.AssetStateCheckedOut) + userAsset, err := tx.BuyAsset(ctx, startsAt, stopsAt, bw, accountID, asset) if err != nil { - return err + return nil, 0, err + } + if err = tx.MergeAssetAt(ctx, assetId, startsAt); err != nil { + return nil, 0, err } - baseSegment := AssetSegment{ - StartsAt: dbAsset.StartAt, - StopsAt: dbAsset.StopsAt, - Bandwidth: dbAsset.Bandwidth, + if err = tx.MergeAssetAt(ctx, assetId, stopsAt); err != nil { + return nil, 0, err } - split, err := SplitAsset(baseSegment, RequestedSplit{ - ExactFrom: asset.StartsAtExactly.AsTime(), - ExactTo: asset.StopsAtExactly.AsTime(), - ExactBandwidth: asset.BandwidthExact, - }) + _, err = tx.RegisterAssetEvent(ctx, userAsset, marketplacedb.AssetBought) if err != nil { - return err + return nil, 0, err } - newAsset := &marketplacedb.DBAsset{ - AccountId: sql.NullInt64{Int64: accountID, Valid: true}, - IA: dbAsset.IA, - BandwidthMin: dbAsset.BandwidthMin, - BandwidthMax: dbAsset.BandwidthMax, - TimeGranularity: dbAsset.TimeGranularity, - TimeMinDuration: dbAsset.TimeMinDuration, - TimeMaxDuration: dbAsset.TimeMaxDuration, - IfIdIngress: dbAsset.IfIdIngress, - IfIdEgress: dbAsset.IfIdEgress, - Bandwidth: split.Split.Bandwidth, - StartAt: split.Split.StartsAt, - StopsAt: split.Split.StopsAt, - Price: dbAsset.Price, + cost, ok := calcCost(userAsset) + if !ok { + return nil, 0, serrors.New("asset parameters would lead to integer overflow") } - id, err := tx.InsertAsset(ctx, newAsset) + _, err = tx.UpdateASMoney(ctx, asset.IA, cost) if err != nil { - return err + return nil, 0, err } - newAsset.ID = id - _, err = tx.RegisterAssetEvent(ctx, newAsset, marketplacedb.AssetBought) + return userAsset, cost, nil + } + for _, asset := range assets { + assetId, err := DatabaseAssetID(asset.AssetId) if err != nil { return err } - totalAssetPrice, err := hbirdmarketplace.ReservationPrice( - dbAsset.Price, - split.Split.Bandwidth, - dbAsset.BandwidthMin, - dbAsset.TimeMinDuration, - dbAsset.TimeGranularity, - split.Split.StopsAt.Sub(split.Split.StartsAt)) + newAsset, cost, err := buyAsset(assetId, asset.StartsAtExactly, asset.StopsAtExactly, asset.BandwidthExact) if err != nil { - return serrors.Wrap("calculating total asset price", err) - } - if totalAssetPrice > math.MaxInt64 { - return serrors.New("total asset price would lead to integer overflow") - } - fee := int64(float64(totalAssetPrice)*float64(s.transactionFeeRelative)) + - int64(s.transactionFeeAbsolute) - pricePlusFee, safe := addInt64(int64(totalAssetPrice), fee) - if !safe { - return serrors.New("total asset price would lead to integer overflow") - } - if _, err := tx.UpdateASMoney(ctx, dbAsset.IA, int64(totalAssetPrice)); err != nil { return err } - costAcc, safe = addInt64(costAcc, pricePlusFee) - if !safe { - return serrors.New("total asset price would lead to integer overflow") - } - boughtAssets = append(boughtAssets, id) - for _, segment := range split.Remainders { - _, err := tx.InsertAsset(ctx, &marketplacedb.DBAsset{ - IA: dbAsset.IA, - BandwidthMin: dbAsset.BandwidthMin, - BandwidthMax: dbAsset.BandwidthMax, - TimeGranularity: dbAsset.TimeGranularity, - TimeMinDuration: dbAsset.TimeMinDuration, - TimeMaxDuration: dbAsset.TimeMaxDuration, - IfIdIngress: dbAsset.IfIdIngress, - IfIdEgress: dbAsset.IfIdEgress, - Bandwidth: segment.Bandwidth, - StartAt: segment.StartsAt, - StopsAt: segment.StopsAt, - Price: dbAsset.Price, - }) - if err != nil { - return err - } + var ok bool + costAcc, ok = addInt64(costAcc, cost) + if !ok { + return serrors.New("asset parameters would lead to integer overflow") } - if err := tx.RemoveAsset(ctx, dbAsset.ID); err != nil { - return err + if uint64(costAcc) > maxPrice { + // we perform this check per asset to be able to terminate early + return serrors.New("assets too expensive") } - } - if uint64(costAcc) > maxPrice { - return serrors.New("cost higher than max price") + boughtIDs = append(boughtIDs, newAsset.ID) } _, err := tx.UpdateAccountMoney(ctx, accountID, -costAcc) - return err + if err != nil { + return err + } + return nil }) if err != nil { return nil, 0, err } - return boughtAssets, costAcc, nil + return boughtIDs, costAcc, nil } func (s *MarketplaceStorage) InsertReservation( @@ -765,7 +822,7 @@ func (s *MarketplaceStorage) UndoRedemption( func validateAsset( a *marketplacedb.DBAsset, ) error { - duration := a.StopsAt.Sub(a.StartAt) + duration := a.StopsAt - a.StartsAt if a.Bandwidth == 0 { return serrors.New("bandwidth is 0") } @@ -777,7 +834,7 @@ func validateAsset( func validateAssetForRedemption( a *marketplacedb.DBAsset, ) error { - duration := a.StopsAt.Sub(a.StartAt) + duration := a.StopsAt - a.StartsAt if err := validateAsset(a); err != nil { return err } @@ -787,13 +844,13 @@ func validateAssetForRedemption( if a.Bandwidth > a.BandwidthMax { return serrors.New("bandwith > max_bandiwdth") } - if duration < time.Duration(a.TimeMinDuration)*time.Second { + if duration < int64(a.TimeMinDuration) { return serrors.New("duration < time_min_duration") } - if duration > time.Duration(a.TimeMaxDuration)*time.Second { + if duration > int64(a.TimeMaxDuration) { return serrors.New("duration > time_max_duration") } - if duration%(time.Duration(a.TimeGranularity)*time.Second) != 0 { + if duration%int64(a.TimeGranularity) != 0 { return serrors.New("duration not multiple of time granularity") } return nil @@ -964,3 +1021,26 @@ func (s *MarketplaceStorage) AssignReservation( } return 1, nil } + +func (s *MarketplaceStorage) LookupAssets(ctx context.Context, ia addr.IA) ([]*marketplacedb.DBAsset, error) { + return s.db.LookupAssets(ctx, ia) +} + +func (s *MarketplaceStorage) LookupAllAssetSegments(ctx context.Context, id int64, ia addr.IA) ([]*marketplacedb.DBAssetSegment, error) { + var segments []*marketplacedb.DBAssetSegment + err := s.db.WithTx(ctx, func(tx marketplacedb.Repository) error { + _, err := tx.LookupAsset(ctx, id, ia) + if err != nil { + return err + } + segments, err = tx.LookupAllAssetSegments(ctx, id) + if err != nil { + return err + } + return nil + }) + if err != nil { + return nil, err + } + return segments, nil +} diff --git a/marketplace/webapp/BUILD.bazel b/marketplace/webapp/BUILD.bazel index ff6b803cd1..024eac8c7f 100644 --- a/marketplace/webapp/BUILD.bazel +++ b/marketplace/webapp/BUILD.bazel @@ -6,7 +6,9 @@ go_library( embedsrcs = [ "static/script.js", "static/style.css", + "static/visualize-assets.js", "templates/account.html", + "templates/asassets.html", "templates/asbalance.html", "templates/aslogin.html", "templates/assets.html", diff --git a/marketplace/webapp/static/style.css b/marketplace/webapp/static/style.css index 6919584167..26856aa6c8 100644 --- a/marketplace/webapp/static/style.css +++ b/marketplace/webapp/static/style.css @@ -286,4 +286,237 @@ select:focus { select option { padding: 8px; +} + + +/* viz section */ +.util { + --util-low: #5fa58f; + --util-mid: #d4a73c; + --util-high: #de7f3f; + --util-full: #cf4a3f; + --util-plot-h: 240px; + + --util-line: color-mix(in srgb, currentColor 18%, transparent); + --util-hatch: color-mix(in srgb, currentColor 25%, transparent); + --util-muted: color-mix(in srgb, currentColor 60%, transparent); +} + +.util *, +.util *::before, +.util *::after { + box-sizing: border-box; +} + +/* bar level colours (shared by bars and legend swatches) */ +.util-low { + --c: var(--util-low); +} + +.util-mid { + --c: var(--util-mid); +} + +.util-high { + --c: var(--util-high); +} + +.util-full { + --c: var(--util-full); +} + +/* header, controls, stats */ +.util-head { + display: flex; + justify-content: space-between; + align-items: center; + gap: 1em; + margin-bottom: 1em; +} + +.util-head h2 { + margin: 0; +} + +.util-error { + margin: 0 0 .75em; + color: var(--util-full); +} + +.util-controls { + display: flex; + flex-wrap: wrap; + align-items: flex-end; + gap: .75em 1em; + margin-bottom: .5em; +} + +.util-controls label { + display: grid; + gap: .2em; + font-size: .8em; + color: var(--util-muted); +} + +.util-buttons { + display: flex; + gap: .35em; +} + +.util-hint { + margin: 0 0 1em; + font-size: .8em; + color: var(--util-muted); +} + +.util-stats { + display: flex; + flex-wrap: wrap; + gap: .25em 1.5em; + margin-bottom: .75em; + font-size: .85em; + color: var(--util-muted); +} + +.util-stats b { + color: inherit; + font-weight: 600; + font-variant-numeric: tabular-nums; +} + +/* chart: y-axis gutter + plot, x-axis underneath */ +.util-chart { + display: grid; + grid-template-columns: 4em 1fr; + margin-top: 30px; +} + +.util-y { + position: relative; + height: var(--util-plot-h); +} + +.util-y span { + position: absolute; + right: .5em; + transform: translateY(50%); + font-size: .72em; + color: var(--util-muted); + font-variant-numeric: tabular-nums; +} + +.util-y span.util-top { + transform: none; +} + +.util-plot { + position: relative; + height: var(--util-plot-h); + cursor: crosshair; + user-select: none; + touch-action: none; + border-left: 1px solid var(--util-line); + border-bottom: 1px solid var(--util-line); +} + +.util-gap { + position: absolute; + top: 0; + bottom: 0; + background: repeating-linear-gradient(135deg, transparent 0 5px, var(--util-hatch) 5px 6px); +} + +/* one slot = full-height hover column with the bar drawn inside it */ +.util-seg { + position: absolute; + top: 0; + bottom: 0; + min-width: 1px; +} + +.util-seg::before { + content: ""; + position: absolute; + left: 0; + right: 0; + bottom: 0; + height: calc(var(--used) * 1%); + background: var(--c); + border-top: 2px solid color-mix(in srgb, var(--c) 70%, #000); +} + +.util-seg:hover { + background: color-mix(in srgb, currentColor 8%, transparent); +} + +.util-hline { + position: absolute; + left: 0; + right: 0; + border-top: 1px dashed var(--util-line); + pointer-events: none; +} + +.util-vline { + position: absolute; + top: 0; + bottom: 0; + border-left: 1px dashed var(--util-line); + pointer-events: none; + opacity: .7; +} + +.util-brush { + position: absolute; + top: 0; + bottom: 0; + pointer-events: none; + background: color-mix(in srgb, currentColor 12%, transparent); + border-left: 1px solid color-mix(in srgb, currentColor 50%, transparent); + border-right: 1px solid color-mix(in srgb, currentColor 50%, transparent); +} + +.util-x { + position: relative; + height: 1.8em; + grid-column: 2; +} + +.util-x span { + position: absolute; + top: .35em; + transform: translateX(-50%); + font-size: .72em; + color: var(--util-muted); + white-space: nowrap; + font-variant-numeric: tabular-nums; +} + +.util-legend { + display: flex; + flex-wrap: wrap; + gap: 1em; + margin-top: .5em; + font-size: .78em; + color: var(--util-muted); +} + +.util-legend i { + display: inline-block; + width: .8em; + height: .8em; + margin-right: .35em; + border-radius: 2px; + vertical-align: -.1em; + background: var(--c); +} + +.util-legend i.util-nodata { + background: repeating-linear-gradient(135deg, transparent 0 3px, var(--util-hatch) 3px 4px); + border: 1px solid var(--util-hatch); +} + +/* optional: highlight for the asset row whose chart is open (class set by the JS) */ +tr.util-active { + background: color-mix(in srgb, currentColor 8%, transparent); } \ No newline at end of file diff --git a/marketplace/webapp/static/visualize-assets.js b/marketplace/webapp/static/visualize-assets.js new file mode 100644 index 0000000000..3e03d8b3aa --- /dev/null +++ b/marketplace/webapp/static/visualize-assets.js @@ -0,0 +1,316 @@ +const Utilization = (() => { + 'use strict'; + + const $ = (id) => document.getElementById(id); + + const MIN_SPAN = 1000; // smallest zoom: 1 second + const MONTHS = ['Jan', 'Feb', 'Mar', 'Apr', 'May', 'Jun', 'Jul', 'Aug', 'Sep', 'Oct', 'Nov', 'Dec']; + + const cache = new Map(); // asset id -> parsed asset + let asset = null; // currently shown asset (null until the first load) + let view = { from: 0, to: 0 }; + + // ---------- formatting ---------- + + const pad = (n) => String(n).padStart(2, '0'); + + function toInput(ms) { + const d = new Date(ms); + return d.getFullYear() + '-' + pad(d.getMonth() + 1) + '-' + pad(d.getDate()) + + 'T' + pad(d.getHours()) + ':' + pad(d.getMinutes()) + ':' + pad(d.getSeconds()); + } + const fmtFull = (ms) => toInput(ms).replace('T', ' '); + const fmtNum = (n) => n.toLocaleString(undefined, { maximumFractionDigits: 1 }); + const level = (pct) => (pct >= 95 ? 'full' : pct >= 75 ? 'high' : pct >= 40 ? 'mid' : 'low'); + const span = () => view.to - view.from; + + // ---------- loading ---------- + + // The page already knows id, bandwidth, start and end (data-* attributes on + // the button). Only the slots are fetched: POST /slots?id= -> [ {starts_at, stops_at, available}, ... ] + // All timestamps from the server are unix seconds; internally everything is milliseconds. + async function load(data) { + if (cache.has(data.id)) return cache.get(data.id); + const res = await fetch('/asassets/data?id=' + encodeURIComponent(data.id), { method: 'POST' }); + if (!res.ok) throw new Error('Could not load utilization (' + res.status + ')'); + const rawSlots = await res.json(); + const a = { + id: data.id, + bandwidth: Number(data.bandwidth), + price: Number(data.price), + start: Number(data.start) * 1000, // unix seconds -> ms for JS Date + end: Number(data.end) * 1000, + slots: rawSlots + .map((s) => ({ s: s.starts_at * 1000, e: s.stops_at * 1000, avail: s.available })) + .sort((p, q) => p.s - q.s), + }; + cache.set(data.id, a); + return a; + } + + function showError(msg) { + const err = $('viz-error'); + err.textContent = msg; + err.hidden = !msg; + } + + function markActiveRow(id) { + document.querySelectorAll('tr[data-id]').forEach((tr) => { + tr.classList.toggle('util-active', tr.dataset.id === id); + }); + } + + // ---------- public actions (called from inline handlers) ---------- + + // onclick on the row button: loads the data, then shows the hidden section. + async function visualize(btn) { + const label = btn.textContent; + btn.disabled = true; + btn.textContent = 'Loading\u2026'; + try { + const a = await load(btn.dataset); + asset = a; + $('viz-title').textContent = "Asset ID " + a.id; + $('stat-bw').textContent = fmtNum(a.bandwidth); + showError(''); + view = { from: a.start, to: a.end }; + $('viz').hidden = false; + markActiveRow(a.id); + render(); + $('viz').scrollIntoView({ behavior: 'smooth', block: 'nearest' }); + } catch (err) { + $('viz').hidden = false; + showError(err.message); + } finally { + btn.disabled = false; + btn.textContent = label; + } + } + + function hide() { + $('viz').hidden = true; + markActiveRow(null); + } + + function setView(from, to) { + if (!asset) return; + const sp = Math.max(to - from, MIN_SPAN); + if (sp >= asset.end - asset.start) { + view = { from: asset.start, to: asset.end }; + } else { + if (from < asset.start) from = asset.start; + if (from + sp > asset.end) from = asset.end - sp; + view = { from, to: from + sp }; + } + render(); + } + + function apply() { + if (!asset) return; + const f = new Date($('from').value).getTime(); + const t = new Date($('to').value).getTime(); + if (Number.isNaN(f) || Number.isNaN(t) || t <= f) { + showError('Enter a valid range: "To" must be after "From".'); + return; + } + showError(''); + setView(f, t); + } + + function zoomIn() { + if (!asset) return; + const c = (view.from + view.to) / 2; + const h = span() / 4; + setView(c - h, c + h); + } + + function zoomOut() { + if (!asset) return; + const c = (view.from + view.to) / 2; + const h = span(); + setView(c - h, c + h); + } + + function panLeft() { + if (asset) setView(view.from - span() / 2, view.to - span() / 2); + } + + function panRight() { + if (asset) setView(view.from + span() / 2, view.to + span() / 2); + } + + function reset() { + if (asset) setView(asset.start, asset.end); + } + + // ---------- drag-to-zoom (pointer handlers on the plot) ---------- + + let dragX0 = null; + let brushEl = null; + + function localX(ev) { + const rect = ev.currentTarget.getBoundingClientRect(); + return { x: Math.min(Math.max(ev.clientX - rect.left, 0), rect.width), width: rect.width }; + } + + function brushStart(ev) { + if (!asset || ev.button !== 0) return; + dragX0 = localX(ev).x; + brushEl = document.createElement('div'); + brushEl.className = 'util-brush'; + brushEl.style.left = dragX0 + 'px'; + brushEl.style.width = '0px'; + ev.currentTarget.appendChild(brushEl); + ev.currentTarget.setPointerCapture(ev.pointerId); + } + + function brushMove(ev) { + if (dragX0 === null) return; + const { x } = localX(ev); + brushEl.style.left = Math.min(x, dragX0) + 'px'; + brushEl.style.width = Math.abs(x - dragX0) + 'px'; + } + + function brushEnd(ev) { + if (dragX0 === null) return; + const { x, width } = localX(ev); + const x0 = dragX0; + brushCancel(); + if (Math.abs(x - x0) > 5) { + const toTime = (px) => view.from + (px / width) * span(); + setView(toTime(Math.min(x, x0)), toTime(Math.max(x, x0))); + } + } + + function brushCancel() { + dragX0 = null; + if (brushEl) { + brushEl.remove(); + brushEl = null; + } + } + + // ---------- axis ticks ---------- + + const S = 1000, M = 60 * S, H = 60 * M, D = 24 * H; + const STEPS = [S, 2 * S, 5 * S, 10 * S, 15 * S, 30 * S, M, 2 * M, 5 * M, 10 * M, 15 * M, 30 * M, + H, 2 * H, 3 * H, 6 * H, 12 * H, D, 2 * D, 7 * D, 14 * D, 30 * D, 90 * D, 365 * D]; + + function makeTicks(from, to) { + const sp = to - from; + const step = STEPS.find((s) => sp / s <= 8) || STEPS[STEPS.length - 1]; + const off = -new Date(from).getTimezoneOffset() * M; // align ticks to local time + const ticks = []; + for (let t = Math.ceil((from + off) / step) * step - off; t <= to; t += step) ticks.push(t); + + const label = (t) => { + const d = new Date(t); + const date = d.getDate() + ' ' + MONTHS[d.getMonth()]; + const hm = pad(d.getHours()) + ':' + pad(d.getMinutes()); + if (step >= D) return date; + const time = step < M ? hm + ':' + pad(d.getSeconds()) : hm; + return sp >= D ? date + ' ' + time : time; + }; + return ticks.map((t) => ({ t, text: label(t) })); + } + + // ---------- rendering ---------- + + function render() { + const { from, to } = view; + const sp = to - from; + const bw = asset.bandwidth; + const pos = (t) => ((t - from) / sp) * 100; + + const plot = $('plot'); + const xAxis = $('x'); + const yAxis = $('y'); + + $('from').value = toInput(from); + $('to').value = toInput(to); + $('from').min = $('to').min = toInput(asset.start); + $('from').max = $('to').max = toInput(asset.end); + + plot.replaceChildren(); + xAxis.replaceChildren(); + yAxis.replaceChildren(); + + // y axis + horizontal grid lines (0 .. bandwidth) + [0, 25, 50, 75, 100].forEach((p) => { + const lbl = document.createElement('span'); + lbl.textContent = fmtNum((bw * p) / 100); + lbl.style.bottom = p + '%'; + if (p === 100) lbl.classList.add('util-top'); + yAxis.appendChild(lbl); + if (p > 0) { + const line = document.createElement('div'); + line.className = 'util-hline'; + line.style.bottom = p + '%'; + plot.appendChild(line); + } + }); + + // x axis + vertical grid lines + makeTicks(from, to).forEach(({ t, text }) => { + const lbl = document.createElement('span'); + lbl.textContent = text; + lbl.style.left = pos(t) + '%'; + xAxis.appendChild(lbl); + const line = document.createElement('div'); + line.className = 'util-vline'; + line.style.left = pos(t) + '%'; + plot.appendChild(line); + }); + + // bars: only slots intersecting the view, clipped to it + const frag = document.createDocumentFragment(); + const addGap = (a, b) => { + const gap = document.createElement('div'); + gap.className = 'util-gap'; + gap.style.left = pos(a) + '%'; + gap.style.width = (pos(b) - pos(a)) + '%'; + gap.title = 'No data\n' + fmtFull(a) + ' \u2013 ' + fmtFull(b); + frag.appendChild(gap); + }; + let cursor = from; // end of the data seen so far; anything after it is a gap + let weighted = 0, covered = 0, peak = 0, n = 0, revenue = 0; + for (const s of asset.slots) { + if (s.e <= from) continue; + if (s.s >= to) break; // slots are sorted by start + const a = Math.max(s.s, from); + const b = Math.min(s.e, to); + if (a > cursor) addGap(cursor, a); + cursor = Math.max(cursor, b); + const sold = Math.min(Math.max(bw - s.avail, 0), bw); + const pct = (sold / bw) * 100; + weighted += pct * (b - a); + covered += b - a; + revenue += sold * asset.price * ((b - a) / 1000) + peak = Math.max(peak, pct); + n++; + + const seg = document.createElement('div'); + seg.className = 'util-seg util-' + level(pct); + seg.style.left = pos(a) + '%'; + seg.style.width = (pos(b) - pos(a)) + '%'; + seg.style.setProperty('--used', pct.toFixed(2)); + seg.title = fmtFull(s.s) + ' \u2013 ' + fmtFull(s.e) + + '\nSold: ' + fmtNum(sold) + ' / ' + fmtNum(bw) + ' (' + pct.toFixed(0) + '%)' + + '\nAvailable: ' + fmtNum(s.avail); + frag.appendChild(seg); + } + if (cursor < to) addGap(cursor, to); + // bars go first in the DOM so the grid lines draw on top of them + plot.insertBefore(frag, plot.firstChild); + + $('stat-avg').textContent = covered ? (weighted / covered).toFixed(0) + '%' : '\u2013'; + $('stat-peak').textContent = n ? peak.toFixed(0) + '%' : '\u2013'; + $('stat-revenue').textContent = n ? fmtNum(Math.floor(revenue)) : '\u2013'; + $('stat-count').textContent = String(n); + } + + return { + visualize, hide, apply, zoomIn, zoomOut, panLeft, panRight, reset, + brushStart, brushMove, brushEnd, brushCancel + }; +})(); diff --git a/marketplace/webapp/templates/asassets.html b/marketplace/webapp/templates/asassets.html new file mode 100644 index 0000000000..c2bbb114f1 --- /dev/null +++ b/marketplace/webapp/templates/asassets.html @@ -0,0 +1,124 @@ + + + + + + View Assets + + + + + + +
+

Assets

+

{{.Error}}

+
+

Assets

+ + + + + + + + + + + + + + {{range $asset := .Assets}} + + + + + + + + + + {{end}} + +
IDIngressEgressStarts atStops atBandwidth
{{.ID}} + {{with .IfIdIngress}} + {{.Value}} + {{end}} + + {{with .IfIdEgress}} + {{.Value}} + {{end}} + {{.StartsAt}}{{.StopsAt}}{{.Bandwidth}} +
+ +
+
+
+ + +
+ + + \ No newline at end of file diff --git a/marketplace/webapp/templates/asbalance.html b/marketplace/webapp/templates/asbalance.html index 6f8cecc438..770eb0d663 100644 --- a/marketplace/webapp/templates/asbalance.html +++ b/marketplace/webapp/templates/asbalance.html @@ -12,6 +12,7 @@