6659027bec
- Worker polls Cloudflare Worker API for pending match jobs - Claims jobs and executes matches using the game engine - Uploads replays to R2 via S3-compatible API - Sends heartbeats during match execution - Submits results back to Worker API - Includes retry logic with exponential backoff - API client tests for job coordination endpoints Also fixes glicko2.ts: export g() and E() functions for testing Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
120 lines
2.9 KiB
Go
120 lines
2.9 KiB
Go
// R2 client for uploading replays
|
|
package main
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"fmt"
|
|
|
|
"github.com/aws/aws-sdk-go-v2/aws"
|
|
"github.com/aws/aws-sdk-go-v2/config"
|
|
"github.com/aws/aws-sdk-go-v2/credentials"
|
|
"github.com/aws/aws-sdk-go-v2/service/s3"
|
|
)
|
|
|
|
// R2Client handles R2 bucket operations.
|
|
type R2Client struct {
|
|
client *s3.Client
|
|
bucket string
|
|
endpoint string
|
|
}
|
|
|
|
// NewR2Client creates a new R2 client.
|
|
func NewR2Client(cfg *Config) *R2Client {
|
|
// Create custom endpoint resolver for R2
|
|
customResolver := aws.EndpointResolverWithOptionsFunc(func(service, region string, options ...interface{}) (aws.Endpoint, error) {
|
|
return aws.Endpoint{
|
|
URL: cfg.R2Endpoint,
|
|
SigningRegion: "auto",
|
|
}, nil
|
|
})
|
|
|
|
// Load AWS config with R2 credentials
|
|
awsCfg, err := config.LoadDefaultConfig(context.TODO(),
|
|
config.WithCredentialsProvider(credentials.NewStaticCredentialsProvider(
|
|
cfg.R2AccessKey,
|
|
cfg.R2SecretKey,
|
|
"",
|
|
)),
|
|
config.WithEndpointResolverWithOptions(customResolver),
|
|
)
|
|
if err != nil {
|
|
panic(fmt.Sprintf("failed to load AWS config: %v", err))
|
|
}
|
|
|
|
return &R2Client{
|
|
client: s3.NewFromConfig(awsCfg),
|
|
bucket: cfg.R2Bucket,
|
|
endpoint: cfg.R2Endpoint,
|
|
}
|
|
}
|
|
|
|
// Upload uploads data to R2.
|
|
func (c *R2Client) Upload(ctx context.Context, key string, data []byte, contentType string) error {
|
|
_, err := c.client.PutObject(ctx, &s3.PutObjectInput{
|
|
Bucket: aws.String(c.bucket),
|
|
Key: aws.String(key),
|
|
Body: bytes.NewReader(data),
|
|
ContentType: aws.String(contentType),
|
|
CacheControl: aws.String("public, max-age=31536000, immutable"),
|
|
})
|
|
return err
|
|
}
|
|
|
|
// Download downloads data from R2.
|
|
func (c *R2Client) Download(ctx context.Context, key string) ([]byte, error) {
|
|
resp, err := c.client.GetObject(ctx, &s3.GetObjectInput{
|
|
Bucket: aws.String(c.bucket),
|
|
Key: aws.String(key),
|
|
})
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer resp.Body.Close()
|
|
|
|
buf := new(bytes.Buffer)
|
|
if _, err := buf.ReadFrom(resp.Body); err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
return buf.Bytes(), nil
|
|
}
|
|
|
|
// Delete deletes an object from R2.
|
|
func (c *R2Client) Delete(ctx context.Context, key string) error {
|
|
_, err := c.client.DeleteObject(ctx, &s3.DeleteObjectInput{
|
|
Bucket: aws.String(c.bucket),
|
|
Key: aws.String(key),
|
|
})
|
|
return err
|
|
}
|
|
|
|
// List lists objects with a prefix.
|
|
func (c *R2Client) List(ctx context.Context, prefix string) ([]string, error) {
|
|
var keys []string
|
|
|
|
paginator := s3.NewListObjectsV2Paginator(c.client, &s3.ListObjectsV2Input{
|
|
Bucket: aws.String(c.bucket),
|
|
Prefix: aws.String(prefix),
|
|
})
|
|
|
|
for paginator.HasMorePages() {
|
|
page, err := paginator.NextPage(ctx)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
for _, obj := range page.Contents {
|
|
if obj.Key != nil {
|
|
keys = append(keys, *obj.Key)
|
|
}
|
|
}
|
|
}
|
|
|
|
return keys, nil
|
|
}
|
|
|
|
// Endpoint returns the R2 endpoint URL.
|
|
func (c *R2Client) Endpoint() string {
|
|
return c.endpoint
|
|
}
|