Compare commits

...
Author SHA1 Message Date
Thomas Camlong 8203ca0c63 fix: add lazyConnect to Redis client to prevent build-time connection errors 2026-05-22 09:31:29 +02:00
Thomas Camlong 5e5492818a test: add tests for Redis lazy-init and startup retry logic 2026-05-22 09:06:40 +02:00
Thomas Camlong a6ec5f789d fix: resolve Redis null client crashes on startup
After #5600 merged tasks/websocket into the Next.js process,
`createRedisConnection()` returns `null` because `CI=true` is baked into
the bundle at build time. This causes `TypeError: Cannot read properties
of null` on every `.publish()` / `.get()` call during cron job startup.

- Remove the CI/DISABLE_REDIS_LOGS null guard from createRedisConnection
- Lazy-init all Redis clients to avoid build-time evaluation
- Add a Redis readiness retry loop before starting cron jobs
2026-05-21 16:33:43 +02:00
ThomasandManuel dade477b32 feat(seed): add demo mode with pre-seeded user (#5739)
Co-authored-by: Manuel <30572287+manuel-rw@users.noreply.github.com>
2026-05-21 16:08:46 +02:00
Thomas 79f88dc3b8 ci: add --affected filtering and fix Turbo cache invalidation (#5740)
* ci: add --affected filtering and fix Turbo cache invalidation

- Remove globalDependencies **/.env which invalidated all caches on any .env change
- Move 40+ env vars from globalEnv to build-specific env (lint/format/typecheck don't need DB_URL etc.)
- Add per-task inputs with test file exclusions so test changes don't trigger build/typecheck
- Enable affectedUsingTaskInputs for task-level --affected detection
- Add --affected to all turbo commands in CI (lint, format, typecheck, build)
- Add fetch-depth: 0 for proper git history comparison
- Add Docker Buildx + GHA layer caching for E2E builds

* ci: add Cloudflare remote cache for Turborepo

Configure turborepo-remote-cache-cloudflare as the remote cache backend.
This commit can be reverted independently to disable remote caching.

* fix(ci): set TURBO_SCM_BASE for --affected to work in PRs

Turbo defaults to comparing against `main`, but PRs target `dev`.
Without explicit base/head refs, --affected falls back to running
all tasks. Use the PR base SHA for pull_requests, the before SHA
for pushes, and HEAD~1 as a last resort.

* perf(typecheck): exclude bundled .cjs/.css from tsc

websocket (835K lines) and tasks (755K lines) had bundled output
files being typechecked due to `include: ["."]` in their tsconfig.
Full typecheck drops from ~70s to ~26s.

* perf(turbo): scope topo task inputs to package.json only

The topo virtual task had no explicit inputs, defaulting to all files.
This caused any source change to cascade cache invalidation through
all dependent packages via ^topo. Since topo is only for dependency
ordering, its hash should only change when dependencies change.

* fix(turbo): disable remote cache signature verification

The cache lives on a private CF worker, signature verification
just adds noise without security benefit.

* fix(turbo): remove runtime env vars from build cache key

These env vars (DB_*, AUTH_*, DOCKER_*, etc.) are all runtime
configuration provided by users when running the Docker container.
They don't influence the build output since Next.js standalone mode
reads process.env at runtime and SKIP_ENV_VALIDATION=true skips
validation during build. Listing them in build.env caused unnecessary
cache busting in local dev.

Also restores coverage report comments per review feedback.

* fix(turbo): include test files in typecheck cache inputs

Test files must be typechecked — excluding them from typecheck inputs
caused changed tests to skip type validation.
2026-05-21 14:11:27 +02:00
16 changed files with 623 additions and 123 deletions

No files matched your search

+2 -1
View File
@@ -10,6 +10,7 @@ README.md
dev
.build
e2e
apps/nextjs/.next
.env
.env.*
!.env.example
!.env.example
+8
View File
@@ -41,6 +41,14 @@ DB_URL='FULL_PATH_TO_YOUR_SQLITE_DB_FILE'
TURBO_TELEMETRY_DISABLED=1
# Turborepo remote cache (optional, for faster CI/local builds)
# Uncomment and fill in to enable remote caching locally.
# CI uses GitHub secrets for these values automatically.
# TURBO_API=https://your-cache-worker.workers.dev
# TURBO_TEAM=team_your_team
# TURBO_TOKEN=your-turbo-token
# TURBO_REMOTE_CACHE_SIGNATURE_KEY=your-signature-key
# Comma-separated paths to Docker-compatible Unix sockets.
# Useful for connecting to multiple runtimes (Docker + Podman) simultaneously.
# When unset and no TCP hosts configured, defaults to /var/run/docker.sock.
+24 -5
View File
@@ -15,6 +15,13 @@ concurrency:
env:
FORCE_COLOR: 3
SKIP_ENV_VALIDATION: true
TURBO_API: ${{ secrets.TURBO_API }}
TURBO_TEAM: ${{ secrets.TURBO_TEAM }}
TURBO_TOKEN: ${{ secrets.TURBO_TOKEN }}
TURBO_REMOTE_CACHE_SIGNATURE_KEY: ${{ secrets.TURBO_REMOTE_CACHE_SIGNATURE_KEY }}
TURBO_SCM_BASE: ${{ github.event.pull_request.base.sha || github.event.before || 'HEAD~1' }}
TURBO_SCM_HEAD: ${{ github.sha }}
jobs:
lint:
@@ -22,6 +29,8 @@ jobs:
timeout-minutes: 10
steps:
- uses: actions/checkout@v6
with:
fetch-depth: 0
- name: Setup
uses: ./tooling/github/setup
@@ -31,31 +40,35 @@ jobs:
run: cp .env.example .env
- name: Lint
run: pnpm lint && pnpm lint:ws
run: pnpm turbo lint --affected && pnpm lint:ws
format:
runs-on: ubuntu-latest
timeout-minutes: 10
steps:
- uses: actions/checkout@v6
with:
fetch-depth: 0
- name: Setup
uses: ./tooling/github/setup
- name: Format
run: pnpm format
run: pnpm turbo format --affected --continue -- --cache --cache-location node_modules/.cache/.prettiercache
typecheck:
runs-on: ubuntu-latest
timeout-minutes: 10
steps:
- uses: actions/checkout@v6
with:
fetch-depth: 0
- name: Setup
uses: ./tooling/github/setup
- name: Typecheck
run: pnpm typecheck
run: pnpm turbo typecheck --affected
test:
runs-on: ubuntu-latest
@@ -78,11 +91,12 @@ jobs:
e2e:
runs-on: ubuntu-latest
timeout-minutes: 15
steps:
- uses: actions/checkout@v6
- name: Setup
uses: ./tooling/github/setup
- name: Set up Docker Buildx
uses: docker/setup-buildx-action@v4
- name: Build docker image
id: build-docker-image
uses: docker/build-push-action@v7
@@ -90,8 +104,11 @@ jobs:
platforms: linux/amd64
context: .
push: false
load: true
tags: homarr-e2e
network: host
cache-from: type=gha
cache-to: type=gha,mode=max
env:
SKIP_ENV_VALIDATION: true
- name: Install playwright browsers
@@ -105,10 +122,12 @@ jobs:
timeout-minutes: 10
steps:
- uses: actions/checkout@v6
with:
fetch-depth: 0
- name: Setup
uses: ./tooling/github/setup
- name: Copy env
shell: bash
run: cp .env.example .env
- name: Build
run: pnpm build
run: pnpm turbo build --affected
+108
View File
@@ -0,0 +1,108 @@
import { afterEach, beforeEach, describe, expect, test, vi } from "vitest";
import { handshakeAsync } from "@homarr/redis";
import { waitForRedisAsync } from "./main";
const { mockLogger } = vi.hoisted(() => ({
mockLogger: {
info: vi.fn(),
warn: vi.fn(),
error: vi.fn(),
debug: vi.fn(),
},
}));
vi.mock("./overrides", () => ({}));
vi.mock("@homarr/core/infrastructure/logs", () => ({
createLogger: () => mockLogger,
}));
vi.mock("@homarr/redis", () => ({
handshakeAsync: vi.fn(),
}));
vi.mock("@homarr/cron-jobs", () => ({
jobGroup: {
initializeAsync: vi.fn().mockResolvedValue(undefined),
startAllAsync: vi.fn().mockResolvedValue(undefined),
},
}));
vi.mock("./on-start", () => ({
onStartAsync: vi.fn().mockResolvedValue(undefined),
}));
describe("waitForRedisAsync", () => {
beforeEach(() => {
vi.clearAllMocks();
vi.useFakeTimers();
});
afterEach(() => {
vi.useRealTimers();
});
test("succeeds on first attempt", async () => {
vi.mocked(handshakeAsync).mockResolvedValueOnce(undefined);
await waitForRedisAsync();
expect(handshakeAsync).toHaveBeenCalledOnce();
expect(mockLogger.info).toHaveBeenCalledWith("Redis connection established");
});
test("retries on failure then succeeds", async () => {
vi.mocked(handshakeAsync)
.mockRejectedValueOnce(new Error("ECONNREFUSED"))
.mockRejectedValueOnce(new Error("ECONNREFUSED"))
.mockResolvedValueOnce(undefined);
const promise = waitForRedisAsync();
await vi.advanceTimersByTimeAsync(500);
await vi.advanceTimersByTimeAsync(500);
await promise;
expect(handshakeAsync).toHaveBeenCalledTimes(3);
expect(mockLogger.warn).toHaveBeenCalledTimes(2);
expect(mockLogger.info).toHaveBeenCalledWith("Redis connection established");
});
test("logs error and proceeds after max retries exhausted", async () => {
vi.mocked(handshakeAsync).mockRejectedValue(new Error("ECONNREFUSED"));
const promise = waitForRedisAsync();
for (let index = 0; index < 10; index++) {
await vi.advanceTimersByTimeAsync(500);
}
await promise;
expect(handshakeAsync).toHaveBeenCalledTimes(10);
expect(mockLogger.error).toHaveBeenCalledWith("Redis did not become ready after retries, proceeding anyway");
expect(mockLogger.info).not.toHaveBeenCalled();
});
test("warns with attempt count on each retry", async () => {
vi.mocked(handshakeAsync).mockRejectedValueOnce(new Error("fail")).mockResolvedValueOnce(undefined);
const promise = waitForRedisAsync();
await vi.advanceTimersByTimeAsync(500);
await promise;
expect(mockLogger.warn).toHaveBeenCalledWith("Redis not ready, retrying (1/10)...");
});
test("does not retry after success", async () => {
vi.mocked(handshakeAsync).mockResolvedValueOnce(undefined);
await waitForRedisAsync();
expect(handshakeAsync).toHaveBeenCalledOnce();
expect(mockLogger.warn).not.toHaveBeenCalled();
expect(mockLogger.error).not.toHaveBeenCalled();
});
});
+22
View File
@@ -1,11 +1,33 @@
// This import has to be the first import in the file so that the agent is overridden before any other modules are imported.
import "./overrides";
import { createLogger } from "@homarr/core/infrastructure/logs";
import { jobGroup } from "@homarr/cron-jobs";
import { handshakeAsync } from "@homarr/redis";
import { onStartAsync } from "./on-start";
const logger = createLogger({ module: "tasks" });
const MAX_RETRIES = 10;
const RETRY_DELAY_MS = 500;
export const waitForRedisAsync = async () => {
for (let attempt = 1; attempt <= MAX_RETRIES; attempt++) {
try {
await handshakeAsync();
logger.info("Redis connection established");
return;
} catch {
logger.warn(`Redis not ready, retrying (${attempt}/${MAX_RETRIES})...`);
await new Promise((resolve) => setTimeout(resolve, RETRY_DELAY_MS));
}
}
logger.error("Redis did not become ready after retries, proceeding anyway");
};
void (async () => {
await waitForRedisAsync();
await onStartAsync();
await jobGroup.initializeAsync();
await jobGroup.startAllAsync();
+1 -1
View File
@@ -9,5 +9,5 @@
"tsBuildInfoFile": "node_modules/.cache/tsbuildinfo.json"
},
"include": ["."],
"exclude": ["node_modules"]
"exclude": ["node_modules", "*.cjs", "*.css"]
}
+1 -1
View File
@@ -9,5 +9,5 @@
"tsBuildInfoFile": "node_modules/.cache/tsbuildinfo.json"
},
"include": ["."],
"exclude": ["node_modules"]
"exclude": ["node_modules", "*.cjs", "*.css"]
}
@@ -5,6 +5,7 @@ import { redisEnv } from "./env";
const defaultRedisOptions = {
connectionName: "homarr",
lazyConnect: true,
} satisfies RedisOptions;
export type { Redis as RedisClient } from "ioredis";
+87 -1
View File
@@ -1,6 +1,7 @@
import { createId, objectKeys } from "@homarr/common";
import {
createDocumentationLink,
credentialsAdminGroup,
everyoneGroup,
getIntegrationDefaultUrl,
getIntegrationName,
@@ -11,12 +12,24 @@ import { defaultServerSettings, defaultServerSettingsKeys } from "@homarr/server
import type { Database } from "..";
import { eq } from "..";
import { getMaxGroupPositionAsync } from "../queries";
import {
getServerSettingByKeyAsync,
insertServerSettingByKeyAsync,
updateServerSettingByKeyAsync,
} from "../queries/server-setting";
import { groups, integrations, onboarding, searchEngines } from "../schema";
import {
boards,
groupMembers,
groupPermissions,
groups,
integrations,
layouts,
onboarding,
searchEngines,
sections,
users,
} from "../schema";
import type { Integration } from "../schema";
export const seedDataAsync = async (db: Database) => {
@@ -25,6 +38,10 @@ export const seedDataAsync = async (db: Database) => {
await seedServerSettingsAsync(db);
await seedDefaultSearchEnginesAsync(db);
await seedDefaultIntegrationsAsync(db);
if (process.env.DEMO_MODE === "true") {
await seedDemoUserAsync(db);
}
};
const seedEveryoneGroupAsync = async (db: Database) => {
@@ -191,3 +208,72 @@ const seedDefaultIntegrationsAsync = async (db: Database) => {
console.log(`Created ${createdCount} default integrations through seeding process`);
};
const seedDemoUserAsync = async (db: Database) => {
const existingUser = await db.query.users.findFirst({
where: eq(users.name, "demo"),
});
if (existingUser) {
console.log("Skipping seeding of demo user as it already exists");
return;
}
const userId = createId();
await db.insert(users).values({
id: userId,
name: "demo",
email: "demo@example.com",
password: "$2b$10$c5AaQb0oj32/bui5Z55PyOb7pY.iEkXYqgdD8MjS9LjSjLMeOkZu2",
});
const maxPosition = await getMaxGroupPositionAsync(db);
const groupId = createId();
await db.insert(groups).values({
id: groupId,
name: credentialsAdminGroup,
ownerId: userId,
position: maxPosition + 1,
});
await db.insert(groupPermissions).values({
groupId,
permission: "admin",
});
await db.insert(groupMembers).values({
groupId,
userId,
});
await db.update(onboarding).set({
step: "finish",
previousStep: "settings",
});
const boardId = createId();
await db.insert(boards).values({
id: boardId,
name: "default",
isPublic: false,
creatorId: userId,
});
await db.insert(sections).values({
id: createId(),
kind: "empty",
xOffset: 0,
yOffset: 0,
boardId,
});
await db.insert(layouts).values({
id: createId(),
name: "Base",
columnCount: 12,
breakpoint: 0,
boardId,
});
await db.update(users).set({ homeBoardId: boardId }).where(eq(users.id, userId));
console.log(
"Your instance has enabled demo mode. Homarr therefore created example data for your demo. You can disable demo mode anytime by setting the env 'DEMO_MODE' to 'false' or unset it.",
);
};
@@ -0,0 +1,102 @@
import { beforeEach, describe, expect, test, vi } from "vitest";
import { ChannelSubscriptionTracker } from "./channel-subscription-tracker";
import { createRedisConnection } from "./connection";
const mockRedisClient = {
subscribe: vi.fn().mockResolvedValue("OK"),
unsubscribe: vi.fn().mockResolvedValue("OK"),
on: vi.fn(),
};
vi.mock("@homarr/core/infrastructure/logs", () => ({
createLogger: () => ({ debug: vi.fn(), warn: vi.fn() }),
}));
vi.mock("./connection", () => ({
createRedisConnection: vi.fn(() => mockRedisClient),
}));
describe("ChannelSubscriptionTracker", () => {
beforeEach(() => {
vi.clearAllMocks();
// Reset static state between tests
// @ts-expect-error accessing private static for test reset
ChannelSubscriptionTracker.subscriptions = new Map();
// @ts-expect-error accessing private static for test reset
ChannelSubscriptionTracker._redis = undefined;
// @ts-expect-error accessing private static for test reset
ChannelSubscriptionTracker.listenerActive = false;
});
test("redis client is lazily created on first subscribe", () => {
vi.mocked(createRedisConnection).mockClear();
ChannelSubscriptionTracker.subscribe("channel-1", vi.fn());
expect(createRedisConnection).toHaveBeenCalledOnce();
});
test("redis client is reused across multiple subscribes", () => {
vi.mocked(createRedisConnection).mockClear();
ChannelSubscriptionTracker.subscribe("channel-1", vi.fn());
ChannelSubscriptionTracker.subscribe("channel-2", vi.fn());
expect(createRedisConnection).toHaveBeenCalledOnce();
});
test("subscribes to redis channel on first callback for a channel", () => {
ChannelSubscriptionTracker.subscribe("test-channel", vi.fn());
expect(mockRedisClient.subscribe).toHaveBeenCalledWith("test-channel");
});
test("does not re-subscribe when adding second callback to same channel", () => {
ChannelSubscriptionTracker.subscribe("test-channel", vi.fn());
mockRedisClient.subscribe.mockClear();
ChannelSubscriptionTracker.subscribe("test-channel", vi.fn());
expect(mockRedisClient.subscribe).not.toHaveBeenCalled();
});
test("unsubscribe function removes the callback", () => {
const unsubscribe = ChannelSubscriptionTracker.subscribe("test-channel", vi.fn());
unsubscribe();
expect(mockRedisClient.unsubscribe).toHaveBeenCalledWith("test-channel");
});
test("does not unsubscribe from redis if other callbacks remain", () => {
ChannelSubscriptionTracker.subscribe("test-channel", vi.fn());
const unsubscribe = ChannelSubscriptionTracker.subscribe("test-channel", vi.fn());
unsubscribe();
expect(mockRedisClient.unsubscribe).not.toHaveBeenCalled();
});
test("activates message listener only once", () => {
ChannelSubscriptionTracker.subscribe("ch-1", vi.fn());
ChannelSubscriptionTracker.subscribe("ch-2", vi.fn());
expect(mockRedisClient.on).toHaveBeenCalledTimes(1);
expect(mockRedisClient.on).toHaveBeenCalledWith("message", expect.any(Function));
});
test("message listener dispatches to correct channel callbacks", () => {
const callback1 = vi.fn();
const callback2 = vi.fn();
ChannelSubscriptionTracker.subscribe("channel-a", callback1);
ChannelSubscriptionTracker.subscribe("channel-b", callback2);
// eslint-disable-next-line @typescript-eslint/no-non-null-assertion
const messageHandler = mockRedisClient.on.mock.calls[0]![1] as (channel: string, message: string) => void;
messageHandler("channel-a", "test-message");
expect(callback1).toHaveBeenCalledWith("test-message");
expect(callback2).not.toHaveBeenCalled();
});
});
@@ -2,6 +2,7 @@ import { randomUUID } from "crypto";
import type { MaybePromise } from "@homarr/common/types";
import { createLogger } from "@homarr/core/infrastructure/logs";
import type { RedisClient } from "@homarr/core/infrastructure/redis";
import { createRedisConnection } from "./connection";
@@ -18,7 +19,10 @@ type SubscriptionCallback = (message: string) => MaybePromise<void>;
*/
export class ChannelSubscriptionTracker {
private static subscriptions = new Map<string, Map<string, SubscriptionCallback>>();
private static redis = createRedisConnection();
private static _redis: RedisClient | undefined;
private static get redis() {
return (this._redis ??= createRedisConnection());
}
private static listenerActive = false;
/**
+147
View File
@@ -0,0 +1,147 @@
import { beforeEach, describe, expect, test, vi } from "vitest";
import {
createCacheChannel,
createGetSetChannel,
createListChannel,
createSubPubChannel,
handshakeAsync,
} from "./channel";
import { createRedisConnection } from "./connection";
const mockRedisClient = {
get: vi.fn().mockResolvedValue(null),
set: vi.fn().mockResolvedValue("OK"),
del: vi.fn().mockResolvedValue(1),
publish: vi.fn().mockResolvedValue(1),
lrange: vi.fn().mockResolvedValue([]),
lrem: vi.fn().mockResolvedValue(0),
lpush: vi.fn().mockResolvedValue(1),
ltrim: vi.fn().mockResolvedValue("OK"),
llen: vi.fn().mockResolvedValue(0),
hello: vi.fn().mockResolvedValue({}),
subscribe: vi.fn().mockResolvedValue("OK"),
on: vi.fn(),
};
vi.mock("@homarr/core/infrastructure/logs", () => ({
createLogger: () => ({ debug: vi.fn(), warn: vi.fn(), error: vi.fn(), info: vi.fn() }),
}));
vi.mock("./connection", () => ({
createRedisConnection: vi.fn(() => mockRedisClient),
}));
vi.mock("./channel-subscription-tracker", () => ({
ChannelSubscriptionTracker: {
subscribe: vi.fn(() => vi.fn()),
},
}));
describe("channel lazy initialization", () => {
beforeEach(() => {
vi.mocked(createRedisConnection).mockClear();
});
test("getSetClient is lazily created on first channel operation", async () => {
vi.mocked(createRedisConnection).mockClear();
const channel = createGetSetChannel<string>("test-lazy");
expect(createRedisConnection).not.toHaveBeenCalled();
await channel.getAsync();
expect(createRedisConnection).toHaveBeenCalled();
});
test("publisher is lazily created on first publish", async () => {
const channel = createSubPubChannel<string>("test-pub", { persist: false });
const callsBefore = vi.mocked(createRedisConnection).mock.calls.length;
await channel.publishAsync("hello");
expect(vi.mocked(createRedisConnection).mock.calls.length).toBeGreaterThan(callsBefore);
});
test("handshakeAsync creates a client and calls hello", async () => {
await handshakeAsync();
expect(mockRedisClient.hello).toHaveBeenCalledOnce();
});
});
describe("createSubPubChannel", () => {
beforeEach(() => {
vi.clearAllMocks();
});
test("publishAsync persists data and publishes", async () => {
const channel = createSubPubChannel<{ value: number }>("test-channel");
await channel.publishAsync({ value: 42 });
expect(mockRedisClient.set).toHaveBeenCalledOnce();
expect(mockRedisClient.publish).toHaveBeenCalledOnce();
});
test("publishAsync with persist:false skips set", async () => {
const channel = createSubPubChannel<string>("no-persist", { persist: false });
await channel.publishAsync("data");
expect(mockRedisClient.set).not.toHaveBeenCalled();
expect(mockRedisClient.publish).toHaveBeenCalledOnce();
});
test("getLastDataAsync returns null when no data", async () => {
const channel = createSubPubChannel<string>("empty");
const result = await channel.getLastDataAsync();
expect(result).toBeNull();
});
});
describe("createListChannel", () => {
beforeEach(() => {
vi.clearAllMocks();
});
test("addAsync pushes to list", async () => {
const channel = createListChannel<string>("test-list");
await channel.addAsync("item1");
expect(mockRedisClient.lpush).toHaveBeenCalledOnce();
});
test("clearAsync deletes the list", async () => {
const channel = createListChannel<string>("test-list");
await channel.clearAsync();
expect(mockRedisClient.del).toHaveBeenCalledOnce();
});
});
describe("createCacheChannel", () => {
beforeEach(() => {
vi.clearAllMocks();
});
test("invalidateAsync deletes the cache key", async () => {
const channel = createCacheChannel<string>("test-cache");
await channel.invalidateAsync();
expect(mockRedisClient.del).toHaveBeenCalledWith("cache:test-cache");
});
test("getAsync returns null when no cached data", async () => {
const channel = createCacheChannel<string>("test-cache");
const result = await channel.getAsync();
expect(result).toBeNull();
});
});
+58 -54
View File
@@ -2,6 +2,7 @@ import superjson from "superjson";
import { createId, hashObjectBase64 } from "@homarr/common";
import { createLogger } from "@homarr/core/infrastructure/logs";
import type { RedisClient } from "@homarr/core/infrastructure/redis";
import type { WidgetKind } from "@homarr/definitions";
import { ChannelSubscriptionTracker } from "./channel-subscription-tracker";
@@ -9,8 +10,11 @@ import { createRedisConnection } from "./connection";
const logger = createLogger({ module: "redisChannel" });
const publisher = createRedisConnection();
const lastDataClient = createRedisConnection();
let _publisher: RedisClient | undefined;
const getPublisher = () => (_publisher ??= createRedisConnection());
let _lastDataClient: RedisClient | undefined;
const getLastDataClient = () => (_lastDataClient ??= createRedisConnection());
/**
* Creates a new pub/sub channel.
@@ -27,34 +31,33 @@ export const createSubPubChannel = <TData>(name: string, { persist }: { persist:
*/
subscribe: (callback: (data: TData) => void) => {
if (persist) {
void lastDataClient.get(lastChannelName).then((data) => {
if (data) {
callback(superjson.parse(data));
}
});
void getLastDataClient()
.get(lastChannelName)
.then((data) => {
if (data) {
callback(superjson.parse(data));
}
});
}
return ChannelSubscriptionTracker.subscribe(channelName, (message) => {
callback(superjson.parse(message));
});
},
/**
* Publish data to the channel with last data saved.
* @param data data to be published
*/
publishAsync: async (data: TData) => {
if (persist) {
await lastDataClient.set(lastChannelName, superjson.stringify(data));
await getLastDataClient().set(lastChannelName, superjson.stringify(data));
}
await publisher.publish(channelName, superjson.stringify(data));
await getPublisher().publish(channelName, superjson.stringify(data));
},
getLastDataAsync: async () => {
const data = await lastDataClient.get(lastChannelName);
const data = await getLastDataClient().get(lastChannelName);
return data ? superjson.parse<TData>(data) : null;
},
};
};
const getSetClient = createRedisConnection();
let _getSetClient: RedisClient | undefined;
const getSetClient = () => (_getSetClient ??= createRedisConnection());
/**
* Creates a new redis channel for a list
@@ -69,7 +72,7 @@ export const createListChannel = <TItem>(name: string) => {
* @returns an array of all items
*/
getAllAsync: async () => {
const items = await getSetClient.lrange(listChannelName, 0, -1);
const items = await getSetClient().lrange(listChannelName, 0, -1);
return items.map((item) => superjson.parse<TItem>(item));
},
/**
@@ -77,20 +80,20 @@ export const createListChannel = <TItem>(name: string) => {
* @param item item to remove
*/
removeAsync: async (item: TItem) => {
await getSetClient.lrem(listChannelName, 0, superjson.stringify(item));
await getSetClient().lrem(listChannelName, 0, superjson.stringify(item));
},
/**
* Clear all items from the channels list
*/
clearAsync: async () => {
await getSetClient.del(listChannelName);
await getSetClient().del(listChannelName);
},
/**
* Add an item to the channels list
* @param item item to add
*/
addAsync: async (item: TItem) => {
await getSetClient.lpush(listChannelName, superjson.stringify(item));
await getSetClient().lpush(listChannelName, superjson.stringify(item));
},
};
};
@@ -106,7 +109,7 @@ export const createGetSetChannel = <TData>(name: string) => {
* @returns data or null if not found
*/
getAsync: async () => {
const data = await getSetClient.get(name);
const data = await getSetClient().get(name);
return data ? superjson.parse<TData>(data) : null;
},
/**
@@ -114,13 +117,13 @@ export const createGetSetChannel = <TData>(name: string) => {
* @param data data to be stored in the channel
*/
setAsync: async (data: TData) => {
await getSetClient.set(name, superjson.stringify(data));
await getSetClient().set(name, superjson.stringify(data));
},
/**
* Remove data from the channel
*/
removeAsync: async () => {
await getSetClient.del(name);
await getSetClient().del(name);
},
};
};
@@ -140,7 +143,7 @@ export const createCacheChannel = <TData>(name: string, cacheDurationMs: number
* @returns data or null if not found or expired
*/
getAsync: async () => {
const data = await getSetClient.get(cacheChannelName);
const data = await getSetClient().get(cacheChannelName);
if (!data) return null;
const parsedData = superjson.parse<{ data: TData; timestamp: Date }>(data);
@@ -156,13 +159,13 @@ export const createCacheChannel = <TData>(name: string, cacheDurationMs: number
* @returns data or new data if not present or expired
*/
consumeAsync: async (callback: () => Promise<TData>) => {
const data = await getSetClient.get(cacheChannelName);
const data = await getSetClient().get(cacheChannelName);
const getNewDataAsync = async () => {
logger.debug(`Cache miss for channel '${cacheChannelName}'`);
const newData = await callback();
const result = { data: newData, timestamp: new Date() };
await getSetClient.set(cacheChannelName, superjson.stringify(result));
await getSetClient().set(cacheChannelName, superjson.stringify(result));
logger.debug(`Cache updated for channel '${cacheChannelName}'`);
return result;
};
@@ -187,14 +190,14 @@ export const createCacheChannel = <TData>(name: string, cacheDurationMs: number
* Invalidate the cache channels data.
*/
invalidateAsync: async () => {
await getSetClient.del(cacheChannelName);
await getSetClient().del(cacheChannelName);
},
/**
* Set the data in the cache channel.
* @param data data to be stored in the cache channel
*/
setAsync: async (data: TData) => {
await getSetClient.set(cacheChannelName, superjson.stringify({ data, timestamp: new Date() }));
await getSetClient().set(cacheChannelName, superjson.stringify({ data, timestamp: new Date() }));
},
};
};
@@ -236,12 +239,12 @@ export const createChannelEventHistory = <TData>(channelName: string, maxElement
});
},
pushAsync: async (data: TData, options = { publish: false }) => {
if (options.publish) await publisher.publish(channelName, superjson.stringify(data));
await getSetClient.lpush(channelName, superjson.stringify({ data, timestamp: new Date() }));
await getSetClient.ltrim(channelName, 0, maxElements);
if (options.publish) await getPublisher().publish(channelName, superjson.stringify(data));
await getSetClient().lpush(channelName, superjson.stringify({ data, timestamp: new Date() }));
await getSetClient().ltrim(channelName, 0, maxElements);
},
clearAsync: async () => {
await getSetClient.del(channelName);
await getSetClient().del(channelName);
},
/**
* Returns a slice of the available data in the channel.
@@ -250,17 +253,17 @@ export const createChannelEventHistory = <TData>(channelName: string, maxElement
* @param endIndex End index of the slice, negative values are counted from the end, defaults at end of range.
*/
getSliceAsync: async (startIndex = 0, endIndex = -1) => {
const range = await getSetClient.lrange(channelName, startIndex, endIndex);
const range = await getSetClient().lrange(channelName, startIndex, endIndex);
return range.map((item) => superjson.parse<{ data: TData; timestamp: Date }>(item));
},
getSliceUntilTimeAsync: async (time: Date) => {
const itemsInCollection = await getSetClient.lrange(channelName, 0, -1);
const itemsInCollection = await getSetClient().lrange(channelName, 0, -1);
return itemsInCollection
.map((item) => superjson.parse<{ data: TData; timestamp: Date }>(item))
.filter((item) => item.timestamp < time);
},
getLengthAsync: async () => {
return await getSetClient.llen(channelName);
return await getSetClient().llen(channelName);
},
name: channelName,
};
@@ -271,11 +274,11 @@ export const createChannelEventHistory = <TData>(channelName: string, maxElement
*/
export const createChannelEventHistoryOld = <TData>(channelName: string, maxElements = 15) => {
const popElementsOverMaxAsync = async () => {
const length = await getSetClient.llen(channelName);
const length = await getSetClient().llen(channelName);
if (length <= maxElements) {
return;
}
await getSetClient.ltrim(channelName, 0, maxElements - 1);
await getSetClient().ltrim(channelName, 0, maxElements - 1);
};
return {
@@ -285,33 +288,33 @@ export const createChannelEventHistoryOld = <TData>(channelName: string, maxElem
});
},
publishAndPushAsync: async (data: TData) => {
await publisher.publish(channelName, superjson.stringify(data));
await getSetClient.lpush(channelName, superjson.stringify({ data, timestamp: new Date() }));
await getPublisher().publish(channelName, superjson.stringify(data));
await getSetClient().lpush(channelName, superjson.stringify({ data, timestamp: new Date() }));
await popElementsOverMaxAsync();
},
pushAsync: async (data: TData) => {
await getSetClient.lpush(channelName, superjson.stringify({ data, timestamp: new Date() }));
await getSetClient().lpush(channelName, superjson.stringify({ data, timestamp: new Date() }));
await popElementsOverMaxAsync();
},
clearAsync: async () => {
await getSetClient.del(channelName);
await getSetClient().del(channelName);
},
getLastAsync: async () => {
const length = await getSetClient.llen(channelName);
const data = await getSetClient.lrange(channelName, length - 1, length);
const length = await getSetClient().llen(channelName);
const data = await getSetClient().lrange(channelName, length - 1, length);
if (data.length !== 1) return null;
// eslint-disable-next-line @typescript-eslint/no-non-null-assertion
return superjson.parse<{ data: TData; timestamp: Date }>(data[0]!);
},
getSliceAsync: async (startIndex: number, endIndex: number) => {
const range = await getSetClient.lrange(channelName, startIndex, endIndex);
const range = await getSetClient().lrange(channelName, startIndex, endIndex);
return range.map((item) => superjson.parse<{ data: TData; timestamp: Date }>(item));
},
getSliceUntilTimeAsync: async (time: Date) => {
const length = await getSetClient.llen(channelName);
const length = await getSetClient().llen(channelName);
const items: TData[] = [];
const itemsInCollection = await getSetClient.lrange(channelName, 0, length - 1);
const itemsInCollection = await getSetClient().lrange(channelName, 0, length - 1);
for (let i = 0; i < length - 1; i++) {
// eslint-disable-next-line @typescript-eslint/no-non-null-assertion
@@ -324,7 +327,7 @@ export const createChannelEventHistoryOld = <TData>(channelName: string, maxElem
return items;
},
getLengthAsync: async () => {
return await getSetClient.llen(channelName);
return await getSetClient().llen(channelName);
},
name: channelName,
};
@@ -338,14 +341,14 @@ export const createChannelWithLatestAndEvents = <TData>(channelName: string) =>
});
},
publishAndUpdateLastStateAsync: async (data: TData) => {
await publisher.publish(channelName, superjson.stringify(data));
await getSetClient.set(channelName, superjson.stringify({ data, timestamp: new Date() }));
await getPublisher().publish(channelName, superjson.stringify(data));
await getSetClient().set(channelName, superjson.stringify({ data, timestamp: new Date() }));
},
setAsync: async (data: TData) => {
await getSetClient.set(channelName, superjson.stringify({ data, timestamp: new Date() }));
await getSetClient().set(channelName, superjson.stringify({ data, timestamp: new Date() }));
},
getAsync: async () => {
const data = await getSetClient.get(channelName);
const data = await getSetClient().get(channelName);
if (!data) return null;
return superjson.parse<{ data: TData; timestamp: Date }>(data);
@@ -354,7 +357,8 @@ export const createChannelWithLatestAndEvents = <TData>(channelName: string) =>
};
};
const queueClient = createRedisConnection();
let _queueClient: RedisClient | undefined;
const getQueueClient = () => (_queueClient ??= createRedisConnection());
type WithId<TItem> = TItem & { _id: string };
@@ -366,11 +370,11 @@ type WithId<TItem> = TItem & { _id: string };
export const createQueueChannel = <TItem>(name: string) => {
const queueChannelName = `queue:${name}`;
const getDataAsync = async () => {
const data = await queueClient.get(queueChannelName);
const data = await getQueueClient().get(queueChannelName);
return data ? superjson.parse<WithId<TItem>[]>(data) : [];
};
const setDataAsync = async (data: WithId<TItem>[]) => {
await queueClient.set(queueChannelName, superjson.stringify(data));
await getQueueClient().set(queueChannelName, superjson.stringify(data));
};
return {
@@ -417,5 +421,5 @@ export const createQueueChannel = <TItem>(name: string) => {
};
export const handshakeAsync = async () => {
await getSetClient.hello();
await getSetClient().hello();
};
+45
View File
@@ -0,0 +1,45 @@
import { describe, expect, test, vi } from "vitest";
import { createRedisClient } from "@homarr/core/infrastructure/redis";
import { createRedisConnection } from "./connection";
const mockRedisInstance = { ping: vi.fn() };
vi.mock("@homarr/core/infrastructure/redis", () => ({
createRedisClient: vi.fn(() => mockRedisInstance),
}));
describe("createRedisConnection", () => {
test("returns a redis client regardless of CI env", () => {
const original = process.env.CI;
process.env.CI = "true";
const client = createRedisConnection();
expect(client).toBe(mockRedisInstance);
expect(client).not.toBeNull();
process.env.CI = original;
});
test("returns a redis client when DISABLE_REDIS_LOGS is set", () => {
const original = process.env.DISABLE_REDIS_LOGS;
process.env.DISABLE_REDIS_LOGS = "true";
const client = createRedisConnection();
expect(client).toBe(mockRedisInstance);
expect(client).not.toBeNull();
process.env.DISABLE_REDIS_LOGS = original;
});
test("always delegates to createRedisClient", () => {
vi.mocked(createRedisClient).mockClear();
createRedisConnection();
expect(createRedisClient).toHaveBeenCalledOnce();
});
});
+1 -13
View File
@@ -1,15 +1,3 @@
import type { RedisClient } from "@homarr/core/infrastructure/redis";
import { createRedisClient } from "@homarr/core/infrastructure/redis";
/**
* Creates a new Redis connection
* @returns redis client
*/
export const createRedisConnection = () => {
if (Boolean(process.env.CI) || Boolean(process.env.DISABLE_REDIS_LOGS)) {
// Return null if we are in CI as we don't want to connect to Redis
return null as unknown as RedisClient;
}
return createRedisClient();
};
export const createRedisConnection = () => createRedisClient();
+11 -46
View File
@@ -1,57 +1,19 @@
{
"$schema": "https://turborepo.org/schema.json",
"globalDependencies": ["**/.env"],
"globalEnv": [
"AUTH_LDAP_BASE",
"AUTH_LDAP_BIND_DN",
"AUTH_LDAP_BIND_PASSWORD",
"AUTH_LDAP_GROUP_CLASS",
"AUTH_LDAP_GROUP_FILTER_EXTRA_ARG",
"AUTH_LDAP_GROUP_MEMBER_ATTRIBUTE",
"AUTH_LDAP_GROUP_MEMBER_USER_ATTRIBUTE",
"AUTH_LDAP_SEARCH_SCOPE",
"AUTH_LDAP_URI",
"AUTH_OIDC_CLIENT_ID",
"AUTH_OIDC_CLIENT_NAME",
"AUTH_OIDC_CLIENT_SECRET",
"AUTH_OIDC_FORCE_USERINFO",
"AUTH_OIDC_ISSUER",
"AUTH_OIDC_SCOPE_OVERWRITE",
"AUTH_OIDC_GROUPS_ATTRIBUTE",
"AUTH_OIDC_NAME_ATTRIBUTE_OVERWRITE",
"AUTH_LDAP_USERNAME_ATTRIBUTE",
"AUTH_LDAP_USER_MAIL_ATTRIBUTE",
"AUTH_LDAP_USERNAME_FILTER_EXTRA_ARG",
"AUTH_OIDC_AUTO_LOGIN",
"AUTH_LOGOUT_REDIRECT_URL",
"AUTH_PROVIDERS",
"AUTH_SESSION_EXPIRY_TIME",
"CI",
"DISABLE_REDIS_LOGS",
"DB_URL",
"DB_HOST",
"DB_USER",
"DB_PASSWORD",
"DB_NAME",
"DB_PORT",
"DB_DRIVER",
"DOCKER_HOSTNAMES",
"DOCKER_PORTS",
"DOCKER_SOCKET_PATHS",
"NODE_ENV",
"PORT",
"LOCAL_CERTIFICATE_PATH",
"LOG_LEVEL",
"SECRET_ENCRYPTION_KEY",
"SKIP_ENV_VALIDATION"
],
"remoteCache": { "signature": false },
"globalEnv": ["CI", "NODE_ENV", "SKIP_ENV_VALIDATION"],
"ui": "stream",
"futureFlags": {
"affectedUsingTaskInputs": true
},
"tasks": {
"topo": {
"dependsOn": ["^topo"]
"dependsOn": ["^topo"],
"inputs": ["package.json"]
},
"build": {
"dependsOn": ["^build"],
"inputs": ["$TURBO_DEFAULT$", "!**/*.test.ts", "!**/*.test.tsx", "!**/*.spec.ts", "!**/__tests__/**"],
"outputs": [".next/**", "!.next/cache/**", "next-env.d.ts", ".output/**", ".vercel/output/**"]
},
"dev": {
@@ -59,15 +21,18 @@
"cache": false
},
"format": {
"inputs": ["$TURBO_DEFAULT$"],
"outputs": ["node_modules/.cache/.prettiercache"],
"outputLogs": "new-only"
},
"lint": {
"dependsOn": ["^topo"],
"inputs": ["$TURBO_DEFAULT$"],
"outputs": ["node_modules/.cache/.eslintcache"]
},
"typecheck": {
"dependsOn": ["^topo"],
"inputs": ["$TURBO_DEFAULT$"],
"outputs": ["node_modules/.cache/tsbuildinfo.json"]
},
"clean": {