push.rocks/smartregistry
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(core): Add pluggable auth providers, storage hooks, multi-upstream cache awareness, and PyPI/RubyGems protocol implementations

Browse Source
This commit is contained in:
Jürgen Kunz
2025-11-27 20:59:49 +00:00
parent 99b01733e7
commit 19da87a9df
12 changed files with 1264 additions and 491 deletions
Download Patch File Download Diff File Expand all files Collapse all files

35
ts/upstream/classes.baseupstream.ts
View File

@@ -110,8 +110,18 @@ export abstract class BaseUpstream {
return null;
}
// Get applicable upstreams sorted by priority
const applicableUpstreams = this.getApplicableUpstreams(context.resource);
if (applicableUpstreams.length === 0) {
return null;
}
// Use the first applicable upstream's URL for cache key
const primaryUpstreamUrl = applicableUpstreams[0]?.url;
// Check cache first
const cached = this.cache.get(context);
const cached = await this.cache.get(context, primaryUpstreamUrl);
if (cached && !cached.stale) {
return {
success: true,
@@ -125,7 +135,7 @@ export abstract class BaseUpstream {
}
// Check for negative cache (recent 404)
if (this.cache.hasNegative(context)) {
if (await this.cache.hasNegative(context, primaryUpstreamUrl)) {
return {
success: false,
status: 404,
@@ -136,13 +146,6 @@ export abstract class BaseUpstream {
};
}
// Get applicable upstreams sorted by priority
const applicableUpstreams = this.getApplicableUpstreams(context.resource);
if (applicableUpstreams.length === 0) {
return null;
}
// If we have stale cache, return it immediately and revalidate in background
if (cached?.stale && this.cacheConfig.staleWhileRevalidate) {
// Fire and forget revalidation
@@ -173,18 +176,19 @@ export abstract class BaseUpstream {
// Cache successful responses
if (result.success && result.body) {
this.cache.set(
await this.cache.set(
context,
Buffer.isBuffer(result.body) ? result.body : Buffer.from(JSON.stringify(result.body)),
result.headers['content-type'] || 'application/octet-stream',
result.headers,
upstream.id,
upstream.url,
);
}
// Cache 404 responses
if (result.status === 404) {
this.cache.setNegative(context, upstream.id);
await this.cache.setNegative(context, upstream.id, upstream.url);
}
return result;
@@ -210,15 +214,15 @@ export abstract class BaseUpstream {
/**
* Invalidate cache for a resource pattern.
*/
public invalidateCache(pattern: RegExp): number {
public async invalidateCache(pattern: RegExp): Promise<number> {
return this.cache.invalidatePattern(pattern);
}
/**
* Clear all cache entries.
*/
public clearCache(): void {
this.cache.clear();
public async clearCache(): Promise<void> {
await this.cache.clear();
}
/**
@@ -501,12 +505,13 @@ export abstract class BaseUpstream {
);
if (result.success && result.body) {
this.cache.set(
await this.cache.set(
context,
Buffer.isBuffer(result.body) ? result.body : Buffer.from(JSON.stringify(result.body)),
result.headers['content-type'] || 'application/octet-stream',
result.headers,
upstream.id,
upstream.url,
);
return; // Successfully revalidated
}

309
ts/upstream/classes.upstreamcache.ts
View File

@@ -4,9 +4,23 @@ import type {
IUpstreamFetchContext,
} from './interfaces.upstream.js';
import { DEFAULT_CACHE_CONFIG } from './interfaces.upstream.js';
import type { IStorageBackend } from '../core/interfaces.core.js';
/**
* In-memory cache for upstream responses.
* Cache metadata stored alongside cache entries.
*/
interface ICacheMetadata {
contentType: string;
headers: Record<string, string>;
cachedAt: string;
expiresAt?: string;
etag?: string;
upstreamId: string;
upstreamUrl: string;
}
/**
* S3-backed upstream cache with in-memory hot layer.
*
* Features:
* - TTL-based expiration
@@ -14,26 +28,45 @@ import { DEFAULT_CACHE_CONFIG } from './interfaces.upstream.js';
* - Negative caching (404s)
* - Content-type aware caching
* - ETag support for conditional requests
* - Multi-upstream support via URL-based cache paths
* - Persistent S3 storage with in-memory hot layer
*
* Note: This is an in-memory implementation. For production with persistence,
* extend this class to use RegistryStorage for S3-backed caching.
* Cache paths are structured as:
* cache/{escaped-upstream-url}/{protocol}:{method}:{path}
*
* @example
* ```typescript
* // In-memory only (default)
* const cache = new UpstreamCache(config);
*
* // With S3 persistence
* const cache = new UpstreamCache(config, 10000, storage);
* ```
*/
export class UpstreamCache {
/** Cache storage */
private readonly cache: Map<string, ICacheEntry> = new Map();
/** In-memory hot cache */
private readonly memoryCache: Map<string, ICacheEntry> = new Map();
/** Configuration */
private readonly config: IUpstreamCacheConfig;
/** Maximum cache entries (prevents memory bloat) */
private readonly maxEntries: number;
/** Maximum in-memory cache entries */
private readonly maxMemoryEntries: number;
/** S3 storage backend (optional) */
private readonly storage?: IStorageBackend;
/** Cleanup interval handle */
private cleanupInterval: ReturnType<typeof setInterval> | null = null;
constructor(config?: Partial<IUpstreamCacheConfig>, maxEntries: number = 10000) {
constructor(
config?: Partial<IUpstreamCacheConfig>,
maxMemoryEntries: number = 10000,
storage?: IStorageBackend
) {
this.config = { ...DEFAULT_CACHE_CONFIG, ...config };
this.maxEntries = maxEntries;
this.maxMemoryEntries = maxMemoryEntries;
this.storage = storage;
// Start periodic cleanup if caching is enabled
if (this.config.enabled) {
@@ -48,17 +81,36 @@ export class UpstreamCache {
return this.config.enabled;
}
/**
* Check if S3 storage is configured.
*/
public hasStorage(): boolean {
return !!this.storage;
}
/**
* Get cached entry for a request context.
* Checks memory first, then falls back to S3.
* Returns null if not found or expired (unless stale-while-revalidate).
*/
public get(context: IUpstreamFetchContext): ICacheEntry | null {
public async get(context: IUpstreamFetchContext, upstreamUrl?: string): Promise<ICacheEntry | null> {
if (!this.config.enabled) {
return null;
}
const key = this.buildCacheKey(context);
const entry = this.cache.get(key);
const key = this.buildCacheKey(context, upstreamUrl);
// Check memory cache first
let entry = this.memoryCache.get(key);
// If not in memory and we have storage, check S3
if (!entry && this.storage) {
entry = await this.loadFromStorage(key);
if (entry) {
// Promote to memory cache
this.memoryCache.set(key, entry);
}
}
if (!entry) {
return null;
@@ -78,7 +130,10 @@ export class UpstreamCache {
}
}
// Entry is too old, remove it
this.cache.delete(key);
this.memoryCache.delete(key);
if (this.storage) {
await this.deleteFromStorage(key).catch(() => {});
}
return null;
}
@@ -86,26 +141,27 @@ export class UpstreamCache {
}
/**
* Store a response in the cache.
* Store a response in the cache (memory and optionally S3).
*/
public set(
public async set(
context: IUpstreamFetchContext,
data: Buffer,
contentType: string,
headers: Record<string, string>,
upstreamId: string,
upstreamUrl: string,
options?: ICacheSetOptions,
): void {
): Promise<void> {
if (!this.config.enabled) {
return;
}
// Enforce max entries limit
if (this.cache.size >= this.maxEntries) {
// Enforce max memory entries limit
if (this.memoryCache.size >= this.maxMemoryEntries) {
this.evictOldest();
}
const key = this.buildCacheKey(context);
const key = this.buildCacheKey(context, upstreamUrl);
const now = new Date();
// Determine TTL based on content type
@@ -122,18 +178,24 @@ export class UpstreamCache {
stale: false,
};
this.cache.set(key, entry);
// Store in memory
this.memoryCache.set(key, entry);
// Store in S3 if available
if (this.storage) {
await this.saveToStorage(key, entry, upstreamUrl).catch(() => {});
}
}
/**
* Store a negative cache entry (404 response).
*/
public setNegative(context: IUpstreamFetchContext, upstreamId: string): void {
public async setNegative(context: IUpstreamFetchContext, upstreamId: string, upstreamUrl: string): Promise<void> {
if (!this.config.enabled || this.config.negativeCacheTtlSeconds <= 0) {
return;
}
const key = this.buildCacheKey(context);
const key = this.buildCacheKey(context, upstreamUrl);
const now = new Date();
const entry: ICacheEntry = {
@@ -146,34 +208,47 @@ export class UpstreamCache {
stale: false,
};
this.cache.set(key, entry);
this.memoryCache.set(key, entry);
if (this.storage) {
await this.saveToStorage(key, entry, upstreamUrl).catch(() => {});
}
}
/**
* Check if there's a negative cache entry for this context.
*/
public hasNegative(context: IUpstreamFetchContext): boolean {
const entry = this.get(context);
public async hasNegative(context: IUpstreamFetchContext, upstreamUrl?: string): Promise<boolean> {
const entry = await this.get(context, upstreamUrl);
return entry !== null && entry.data.length === 0;
}
/**
* Invalidate a specific cache entry.
*/
public invalidate(context: IUpstreamFetchContext): boolean {
const key = this.buildCacheKey(context);
return this.cache.delete(key);
public async invalidate(context: IUpstreamFetchContext, upstreamUrl?: string): Promise<boolean> {
const key = this.buildCacheKey(context, upstreamUrl);
const deleted = this.memoryCache.delete(key);
if (this.storage) {
await this.deleteFromStorage(key).catch(() => {});
}
return deleted;
}
/**
* Invalidate all entries matching a pattern.
* Useful for invalidating all versions of a package.
*/
public invalidatePattern(pattern: RegExp): number {
public async invalidatePattern(pattern: RegExp): Promise<number> {
let count = 0;
for (const key of this.cache.keys()) {
for (const key of this.memoryCache.keys()) {
if (pattern.test(key)) {
this.cache.delete(key);
this.memoryCache.delete(key);
if (this.storage) {
await this.deleteFromStorage(key).catch(() => {});
}
count++;
}
}
@@ -183,11 +258,14 @@ export class UpstreamCache {
/**
* Invalidate all entries from a specific upstream.
*/
public invalidateUpstream(upstreamId: string): number {
public async invalidateUpstream(upstreamId: string): Promise<number> {
let count = 0;
for (const [key, entry] of this.cache.entries()) {
for (const [key, entry] of this.memoryCache.entries()) {
if (entry.upstreamId === upstreamId) {
this.cache.delete(key);
this.memoryCache.delete(key);
if (this.storage) {
await this.deleteFromStorage(key).catch(() => {});
}
count++;
}
}
@@ -195,10 +273,13 @@ export class UpstreamCache {
}
/**
* Clear all cache entries.
* Clear all cache entries (memory and S3).
*/
public clear(): void {
this.cache.clear();
public async clear(): Promise<void> {
this.memoryCache.clear();
// Note: S3 cleanup would require listing and deleting all cache/* objects
// This is left as a future enhancement for bulk cleanup
}
/**
@@ -211,7 +292,7 @@ export class UpstreamCache {
let totalSize = 0;
const now = new Date();
for (const entry of this.cache.values()) {
for (const entry of this.memoryCache.values()) {
totalSize += entry.data.length;
if (entry.data.length === 0) {
@@ -224,13 +305,14 @@ export class UpstreamCache {
}
return {
totalEntries: this.cache.size,
totalEntries: this.memoryCache.size,
freshEntries: freshCount,
staleEntries: staleCount,
negativeEntries: negativeCount,
totalSizeBytes: totalSize,
maxEntries: this.maxEntries,
maxEntries: this.maxMemoryEntries,
enabled: this.config.enabled,
hasStorage: !!this.storage,
};
}
@@ -244,17 +326,136 @@ export class UpstreamCache {
}
}
// ========================================================================
// Storage Methods
// ========================================================================
/**
* Build storage path for a cache key.
* Escapes upstream URL for safe use in S3 paths.
*/
private buildStoragePath(key: string): string {
return `cache/${key}`;
}
/**
* Build storage path for cache metadata.
*/
private buildMetadataPath(key: string): string {
return `cache/${key}.meta`;
}
/**
* Load a cache entry from S3 storage.
*/
private async loadFromStorage(key: string): Promise<ICacheEntry | null> {
if (!this.storage) return null;
try {
const dataPath = this.buildStoragePath(key);
const metaPath = this.buildMetadataPath(key);
// Load data and metadata in parallel
const [data, metaBuffer] = await Promise.all([
this.storage.getObject(dataPath),
this.storage.getObject(metaPath),
]);
if (!data || !metaBuffer) {
return null;
}
const meta: ICacheMetadata = JSON.parse(metaBuffer.toString('utf-8'));
return {
data,
contentType: meta.contentType,
headers: meta.headers,
cachedAt: new Date(meta.cachedAt),
expiresAt: meta.expiresAt ? new Date(meta.expiresAt) : undefined,
etag: meta.etag,
upstreamId: meta.upstreamId,
stale: false,
};
} catch {
return null;
}
}
/**
* Save a cache entry to S3 storage.
*/
private async saveToStorage(key: string, entry: ICacheEntry, upstreamUrl: string): Promise<void> {
if (!this.storage) return;
const dataPath = this.buildStoragePath(key);
const metaPath = this.buildMetadataPath(key);
const meta: ICacheMetadata = {
contentType: entry.contentType,
headers: entry.headers,
cachedAt: entry.cachedAt.toISOString(),
expiresAt: entry.expiresAt?.toISOString(),
etag: entry.etag,
upstreamId: entry.upstreamId,
upstreamUrl,
};
// Save data and metadata in parallel
await Promise.all([
this.storage.putObject(dataPath, entry.data),
this.storage.putObject(metaPath, Buffer.from(JSON.stringify(meta), 'utf-8')),
]);
}
/**
* Delete a cache entry from S3 storage.
*/
private async deleteFromStorage(key: string): Promise<void> {
if (!this.storage) return;
const dataPath = this.buildStoragePath(key);
const metaPath = this.buildMetadataPath(key);
await Promise.all([
this.storage.deleteObject(dataPath).catch(() => {}),
this.storage.deleteObject(metaPath).catch(() => {}),
]);
}
// ========================================================================
// Helper Methods
// ========================================================================
/**
* Escape a URL for safe use in storage paths.
*/
private escapeUrl(url: string): string {
// Remove protocol prefix and escape special characters
return url
.replace(/^https?:\/\//, '')
.replace(/[\/\\:*?"<>|]/g, '_')
.replace(/__+/g, '_');
}
/**
* Build a unique cache key for a request context.
* Includes escaped upstream URL for multi-upstream support.
*/
private buildCacheKey(context: IUpstreamFetchContext): string {
private buildCacheKey(context: IUpstreamFetchContext, upstreamUrl?: string): string {
// Include method, protocol, path, and sorted query params
const queryString = Object.keys(context.query)
.sort()
.map(k => `${k}=${context.query[k]}`)
.join('&');
return `${context.protocol}:${context.method}:${context.path}${queryString ? '?' + queryString : ''}`;
const baseKey = `${context.protocol}:${context.method}:${context.path}${queryString ? '?' + queryString : ''}`;
if (upstreamUrl) {
return `${this.escapeUrl(upstreamUrl)}/${baseKey}`;
}
return baseKey;
}
/**
@@ -333,27 +534,27 @@ export class UpstreamCache {
*/
private evictOldest(): void {
// Evict 10% of max entries
const evictCount = Math.ceil(this.maxEntries * 0.1);
const evictCount = Math.ceil(this.maxMemoryEntries * 0.1);
let evicted = 0;
// First, try to evict stale entries
const now = new Date();
for (const [key, entry] of this.cache.entries()) {
for (const [key, entry] of this.memoryCache.entries()) {
if (evicted >= evictCount) break;
if (entry.stale || (entry.expiresAt && entry.expiresAt < now)) {
this.cache.delete(key);
this.memoryCache.delete(key);
evicted++;
}
}
// If not enough evicted, evict oldest by cachedAt
if (evicted < evictCount) {
const entries = Array.from(this.cache.entries())
const entries = Array.from(this.memoryCache.entries())
.sort((a, b) => a[1].cachedAt.getTime() - b[1].cachedAt.getTime());
for (const [key] of entries) {
if (evicted >= evictCount) break;
this.cache.delete(key);
this.memoryCache.delete(key);
evicted++;
}
}
@@ -375,17 +576,17 @@ export class UpstreamCache {
}
/**
* Remove all expired entries.
* Remove all expired entries from memory cache.
*/
private cleanup(): void {
const now = new Date();
const staleDeadline = new Date(now.getTime() - this.config.staleMaxAgeSeconds * 1000);
for (const [key, entry] of this.cache.entries()) {
for (const [key, entry] of this.memoryCache.entries()) {
if (entry.expiresAt) {
// Remove if past stale deadline
if (entry.expiresAt < staleDeadline) {
this.cache.delete(key);
this.memoryCache.delete(key);
}
}
}
@@ -406,7 +607,7 @@ export interface ICacheSetOptions {
* Cache statistics.
*/
export interface ICacheStats {
/** Total number of cached entries */
/** Total number of cached entries in memory */
totalEntries: number;
/** Number of fresh (non-expired) entries */
freshEntries: number;
@@ -414,10 +615,12 @@ export interface ICacheStats {
staleEntries: number;
/** Number of negative cache entries */
negativeEntries: number;
/** Total size of cached data in bytes */
/** Total size of cached data in bytes (memory only) */
totalSizeBytes: number;
/** Maximum allowed entries */
/** Maximum allowed memory entries */
maxEntries: number;
/** Whether caching is enabled */
enabled: boolean;
/** Whether S3 storage is configured */
hasStorage: boolean;
}