s3 based cache

AWS.js

const _ = require('lodash');
const AWS = require('aws-sdk');
const rx = require('rxjs');
const rxop = require('rxjs/operators');

AWS.config.update({
    accessKeyId: process.env.ACCESS_KEY_ID,
    secretAccessKey: process.env.SECRET_ACCESS_KEY,
    region: process.env.REGION
});

class Client {
    constructor() {
        this.clients = {
            s3: new AWS.S3()
        };

        this.s3 = {
            clear: args => {
                return this.s3.list({
                        prefix: args.prefix
                    })
                    .pipe(
                        rxop.mergeMap(keys => {
                            if(!_.size(keys)) {
                                return rx.of(null);
                            }

                            return this.s3.delete({
                                keys
                            });
                        })
                    );
            },
            delete: args => {
                return new rx.Observable(subscriber => {
                    this.clients.s3.deleteObjects({
                        Bucket: config.aws.bucket,
                        Delete: {
                            Objects: _.map(args.keys, key => {
                                return {
                                    Key: _.trim(key, '/')
                                };
                            })
                        }
                    }, (err, data) => {
                        if (err) {
                            return subscriber.error(err);
                        }

                        subscriber.next(data);
                        subscriber.complete();
                    });
                });
            },
            get: args => {
                return new rx.Observable(subscriber => {
                    this.clients.s3.getObject({
                        Bucket: config.aws.bucket,
                        Key: _.trim(args.key, '/')
                    }, (err, data) => {
                        if (err && err.name !== 'NoSuchKey') {
                            return subscriber.error(err);
                        } else if (err) {
                            subscriber.next({
                                body: null,
                                lastModified: 0
                            });
                        } else {
                            subscriber.next({
                                body: data.Body,
                                lastModified: new Date(data.LastModified).getTime()
                            });
                        }

                        subscriber.complete();
                    });
                });
            },
            list: args => {
                return new rx.Observable(subscriber => {
                    this.clients.s3.listObjects({
                        Bucket: config.aws.bucket,
                        Prefix: _.trim(args.prefix, '/')
                    }, (err, data) => {
                        if (err) {
                            return subscriber.error(err);
                        }

                        subscriber.next(_.map(data.Contents, 'Key'));
                        subscriber.complete();
                    });
                });
            },
            put: args => {
                return new rx.Observable(subscriber => {
                    this.clients.s3.putObject({
                        ACL: 'public-read',
                        Body: args.body,
                        Bucket: config.aws.bucket,
                        CacheControl: args.cacheControl,
                        ContentType: args.contentType,
                        Key: _.trim(args.key, '/')
                    }, (err, data) => {
                        if (err) {
                            return subscriber.error(err);
                        }

                        subscriber.next(data);
                        subscriber.complete();
                    });
                });
            }
        };
    }
}

module.exports = new Client();

index.js

const CacheDriver = require('rxjs-cache-driver');
const S3 = require('./S3');

const s3 = new S3();

const cache = new CacheDriver({
    gzip: true,
    clear: s3.clear.bind(s3),
    del: s3.delete.bind(s3),
    get: s3.get.bind(s3),
    set: s3.set.bind(s3),
    ttr: 7200 //time to refresh, after this will serve cached and refresh is background, so it always serves fast
});

cache.get({
    namespace: 'items',
    id: 'item-one'
}, () => {
    return this.get(); // how to get if is missing on cache, returns a rxjs
});

S3.js

const rxop = require('rxjs/operators');
const AWS = require('./AWS');

module.exports = class {
    clear(args) {
        return AWS.s3.clear({
            prefix: `/cache/${args.namespace}`
        });
    }

    delete(args) {
        return AWS.s3.delete({
            keys: [
                args.id ? `/cache/${args.namespace}/${args.id}` : `/cache/${args.namespace}`
            ]
        });
    }

    get(args) {
        return AWS.s3.get({
                key: args.id ? `/cache/${args.namespace}/${args.id}` : `/cache/${args.namespace}`
            })
            .pipe(
                rxop.map(response => {
                    if (!response.body) {
                        return null;
                    }

                    return {
                        createdAt: response.lastModified,
                        id: args.id || null,
                        namespace: args.namespace,
                        value: response.body
                    };
                })
            );
    }

    set(args) {
        return AWS.s3.put({
            body: args.value,
            cacheControl: 'no-cache',
            contentType: 'application/tar+gzip',
            key: args.id ? `/cache/${args.namespace}/${args.id}` : `/cache/${args.namespace}`
        });
    }
};