maddy/matrix
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Query migration to db done - Query persistence in cache is also done

Browse Source
This commit is contained in:
Narayanan Madaswamy
2026-08-01 16:08:52 +05:30
parent 934937feb0
commit 0dae53017d
26 changed files with 608 additions and 1564 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
.vscode/launch.json vendored
View File

@@ -59,6 +59,7 @@
"REDIS_HOST": "103.125.129.116",
"REDIS_PORT": "7901",
"REDIS_PASSWORD": "M@triXR3d1s@6202",
"REDIS_DATABASE": "1",
"REDIS_SSL": "false",
"CYGNUS_CLOUD_BASE_URL": "http://localhost:8090",
"CYGNUS_TOKEN_URL": "http://localhost:8090/oauth2/token",

4
cygnus-cloud-client/src/main/java/com/cygnus/client/CloudIdentityClient.java
View File

@@ -52,10 +52,10 @@ public class CloudIdentityClient {
.timeout(properties.requestTimeout());
}
public Mono<CloudQueryResponse> fetchQuery(String queryId) {
public Mono<CloudQueryResponse> fetchQuery(int queryId) {
return tokenProvider.accessToken()
.flatMap(token -> webClient.get()
.uri(properties.baseUri().resolve("/api/v1/queries/").resolve(queryId))
.uri(properties.baseUri().resolve("/api/v1/queries/" + queryId))
.header(HttpHeaders.AUTHORIZATION, "Bearer " + token)
.accept(MediaType.APPLICATION_JSON)
.retrieve()

3
cygnus-cloud-client/src/main/java/com/cygnus/client/model/CloudQueryResponse.java
View File

@@ -1,5 +1,4 @@
package com.cygnus.client.model;
public record CloudQueryResponse(String queryId, String query) {
public record CloudQueryResponse(int queryId, String query) {
}

3
cygnus-cloud-service/src/main/java/com/cygnus/cloud/query/CloudQuery.java
View File

@@ -1,5 +1,4 @@
package com.cygnus.cloud.query;
public record CloudQuery(String queryId, String query) {
public record CloudQuery(int queryId, String query) {
}

6
cygnus-cloud-service/src/main/java/com/cygnus/cloud/query/CloudQueryController.java
View File

@@ -1,6 +1,6 @@
package com.cygnus.cloud.query;
import jakarta.validation.constraints.Pattern;
import jakarta.validation.constraints.Min;
import org.springframework.validation.annotation.Validated;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
@@ -21,10 +21,8 @@ public class CloudQueryController {
@GetMapping("/{queryId}")
public Mono<CloudQuery> query(
@PathVariable
@Pattern(regexp = "^[A-Za-z0-9._-]+$") String queryId) {
@PathVariable @Min(1) int queryId) {
return repository.findEnabled(queryId)
.switchIfEmpty(Mono.error(new QueryNotFoundException(queryId)));
}
}

57
cygnus-cloud-service/src/main/java/com/cygnus/cloud/query/QueryCatalogInitializer.java
View File

@@ -1,57 +0,0 @@
package com.cygnus.cloud.query;
import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.core.io.ClassPathResource;
import org.springframework.stereotype.Component;
import reactor.core.publisher.Flux;
@Component
public class QueryCatalogInitializer implements ApplicationRunner {
static final String DELIMITER = "!C0L!";
private final QueryCatalogRepository repository;
public QueryCatalogInitializer(QueryCatalogRepository repository) {
this.repository = repository;
}
@Override
public void run(ApplicationArguments arguments) {
List<CloudQuery> queries = readCatalog();
repository.initialize()
.thenMany(Flux.fromIterable(queries)
.concatMap(repository::insertIfAbsent))
.then()
.block(Duration.ofMinutes(2));
}
List<CloudQuery> readCatalog() {
ClassPathResource resource = new ClassPathResource("queries/nimble.qry");
List<CloudQuery> queries = new ArrayList<>();
try (BufferedReader reader = new BufferedReader(new InputStreamReader(
resource.getInputStream(), StandardCharsets.UTF_8))) {
String line;
while ((line = reader.readLine()) != null) {
int delimiter = line.indexOf(DELIMITER);
if (delimiter <= 0) {
continue;
}
String queryId = line.substring(0, delimiter).trim();
String query = line.substring(delimiter + DELIMITER.length());
if (queryId.matches("^Query\\d+$") && !query.isBlank()) {
queries.add(new CloudQuery(queryId, query));
}
}
return List.copyOf(queries);
} catch (Exception exception) {
throw new IllegalStateException("Unable to load cloud query catalog", exception);
}
}
}

21
cygnus-cloud-service/src/main/java/com/cygnus/cloud/query/QueryCatalogRepository.java
View File

@@ -11,13 +11,13 @@ public class QueryCatalogRepository {
private static final String INITIALIZE = """
CREATE SCHEMA IF NOT EXISTS platform;
CREATE TABLE IF NOT EXISTS platform.application_query (
query_id varchar(100) PRIMARY KEY,
query_id integer PRIMARY KEY,
query_text text NOT NULL,
enabled boolean NOT NULL DEFAULT true,
created_at timestamptz NOT NULL DEFAULT current_timestamp,
updated_at timestamptz NOT NULL DEFAULT current_timestamp,
CONSTRAINT ck_platform_application_query_id
CHECK (query_id ~ '^[A-Za-z0-9._-]+$'),
CHECK (query_id > 0),
CONSTRAINT ck_platform_application_query_text
CHECK (length(btrim(query_text)) > 0)
)
@@ -28,12 +28,6 @@ public class QueryCatalogRepository {
WHERE query_id = $1
AND enabled = true
""";
private static final String INSERT_IF_ABSENT = """
INSERT INTO platform.application_query (query_id, query_text)
VALUES ($1, $2)
ON CONFLICT (query_id) DO NOTHING
""";
private final ReactiveDatabaseClient database;
public QueryCatalogRepository(ReactiveDatabaseClient database) {
@@ -44,18 +38,11 @@ public class QueryCatalogRepository {
return database.query(INITIALIZE).then();
}
public Mono<CloudQuery> findEnabled(String queryId) {
public Mono<CloudQuery> findEnabled(int queryId) {
return database.preparedQuery(FIND, Tuple.of(queryId))
.flatMapMany(rows -> reactor.core.publisher.Flux.fromIterable(rows))
.next()
.map(row -> new CloudQuery(
row.getString("query_id"), row.getString("query_text")));
}
public Mono<Void> insertIfAbsent(CloudQuery query) {
return database.preparedUpdate(
INSERT_IF_ABSENT,
Tuple.of(query.queryId(), query.query()))
.then();
row.getInteger("query_id"), row.getString("query_text")));
}
}

22
cygnus-cloud-service/src/main/java/com/cygnus/cloud/query/QueryCatalogSchemaInitializer.java Normal file
View File

@@ -0,0 +1,22 @@
package com.cygnus.cloud.query;
import java.time.Duration;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.stereotype.Component;
@Component
public class QueryCatalogSchemaInitializer implements ApplicationRunner {
private final QueryCatalogRepository repository;
public QueryCatalogSchemaInitializer(QueryCatalogRepository repository) {
this.repository = repository;
}
@Override
public void run(ApplicationArguments arguments) {
repository.initialize()
.block(Duration.ofMinutes(2));
}
}

3
cygnus-cloud-service/src/main/java/com/cygnus/cloud/query/QueryNotFoundException.java
View File

@@ -6,8 +6,7 @@ import org.springframework.web.bind.annotation.ResponseStatus;
@ResponseStatus(HttpStatus.NOT_FOUND)
public class QueryNotFoundException extends RuntimeException {
public QueryNotFoundException(String queryId) {
public QueryNotFoundException(int queryId) {
super("Query was not found: " + queryId);
}
}

5
cygnus-cloud-service/src/main/resources/db/platform/001_query_catalog.sql
View File

@@ -1,14 +1,13 @@
CREATE SCHEMA IF NOT EXISTS platform;
CREATE TABLE IF NOT EXISTS platform.application_query (
query_id varchar(100) PRIMARY KEY,
query_id integer PRIMARY KEY,
query_text text NOT NULL,
enabled boolean NOT NULL DEFAULT true,
created_at timestamptz NOT NULL DEFAULT current_timestamp,
updated_at timestamptz NOT NULL DEFAULT current_timestamp,
CONSTRAINT ck_platform_application_query_id
CHECK (query_id ~ '^[A-Za-z0-9._-]+$'),
CHECK (query_id > 0),
CONSTRAINT ck_platform_application_query_text
CHECK (length(btrim(query_text)) > 0)
);

25
cygnus-cloud-service/src/main/resources/db/platform/002_numeric_query_ids.sql Normal file
View File

@@ -0,0 +1,25 @@
DO $$
DECLARE
current_type text;
BEGIN
SELECT data_type
INTO current_type
FROM information_schema.columns
WHERE table_schema = 'platform'
AND table_name = 'application_query'
AND column_name = 'query_id';
IF current_type IN ('character varying', 'text') THEN
ALTER TABLE platform.application_query
DROP CONSTRAINT IF EXISTS ck_platform_application_query_id;
ALTER TABLE platform.application_query
ALTER COLUMN query_id TYPE integer
USING regexp_replace(query_id, '^Query', '')::integer;
ALTER TABLE platform.application_query
ADD CONSTRAINT ck_platform_application_query_id
CHECK (query_id > 0);
END IF;
END
$$;

514
cygnus-cloud-service/src/main/resources/queries/nimble.qry
View File

File diff suppressed because one or more lines are too long

20
cygnus-cloud-service/src/test/java/com/cygnus/cloud/query/QueryCatalogInitializerTest.java
View File

@@ -1,20 +0,0 @@
package com.cygnus.cloud.query;
import static org.assertj.core.api.Assertions.assertThat;
import org.junit.jupiter.api.Test;
class QueryCatalogInitializerTest {
@Test
void readsUniqueNumericQueryIdsFromCloudCatalog() {
var queries = new QueryCatalogInitializer(null).readCatalog();
assertThat(queries).hasSize(455);
assertThat(queries).extracting(CloudQuery::queryId).doesNotHaveDuplicates();
assertThat(queries).anySatisfy(query -> {
assertThat(query.queryId()).isEqualTo("Query3");
assertThat(query.query()).startsWith("select!C0L!select portfolio_id");
});
}
}

1
cygnus-installer/src/main/java/com/cygnus/installer/DeploymentWriter.java
View File

@@ -192,6 +192,7 @@ public class DeploymentWriter {
appendEnvironment(env, "REDIS_HOST", runtime.redisHost());
appendEnvironment(env, "REDIS_PORT", runtime.redisPort());
appendEnvironment(env, "REDIS_PASSWORD", runtime.redisPassword());
appendEnvironment(env, "REDIS_DATABASE", "1");
appendEnvironment(env, "REDIS_SSL", Boolean.toString(runtime.redisSsl()));
appendEnvironment(
env, "CYGNUS_CLOUD_BASE_URL", normalizedCloudServiceUrl(runtime));

1
cygnus-installer/src/test/java/com/cygnus/installer/DeploymentWriterTest.java
View File

@@ -109,6 +109,7 @@ class DeploymentWriterTest {
.contains("MATRIX_IMAGE=\"registry.example.com/matrix:1.0.0\"")
.contains("MATRIX_DB_URL=\"jdbc:postgresql://db:5432/matrix\"")
.contains("REDIS_HOST=\"redis\"")
.contains("REDIS_DATABASE=\"1\"")
.contains("CYGNUS_CLOUD_BASE_URL=\"http://host.docker.internal:8090\"")
.contains("CYGNUS_TOKEN_URL=\"http://host.docker.internal:8090/oauth2/token\"")
.contains("CYGNUS_CLIENT_ID=\"matrix-client\"")

388
cygnus-onprem-app/build/classes/matrix/nimble/conf/nimble.qry
View File

File diff suppressed because one or more lines are too long

10
cygnus-onprem-app/pom.xml
View File

@@ -173,16 +173,6 @@
<build>
<finalName>matrix</finalName>
<resources>
<resource>
<directory>src/main/resources</directory>
<excludes>
<!-- Queries are supplied by the cloud query catalog at runtime. -->
<exclude>matrix/nimble/conf/nimble.qry</exclude>
</excludes>
</resource>
</resources>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>

5
cygnus-onprem-app/src/main/java/matrix/nimble/cloud/cache/OnPremRedisConfiguration.java vendored
View File

@@ -16,11 +16,16 @@ public class OnPremRedisConfiguration {
@Value("${REDIS_HOST:192.168.0.111}") String host,
@Value("${REDIS_PORT:7901}") int port,
@Value("${REDIS_PASSWORD:}") String password,
@Value("${REDIS_DATABASE:1}") int database,
@Value("${REDIS_SSL:false}") boolean ssl,
@Value("${REDIS_CONNECT_TIMEOUT_SECONDS:3}") long connectTimeoutSeconds) {
if (database < 0) {
throw new IllegalArgumentException("REDIS_DATABASE must be zero or greater");
}
RedisURI.Builder uri = RedisURI.builder()
.withHost(host)
.withPort(port)
.withDatabase(database)
.withSsl(ssl)
.withTimeout(Duration.ofSeconds(connectTimeoutSeconds));
if (password != null && !password.isBlank()) {

3
cygnus-onprem-app/src/main/java/matrix/nimble/query/CloudQuerySource.java
View File

@@ -11,7 +11,7 @@ public final class CloudQuerySource implements QuerySource {
}
@Override
public String fetch(String queryId) {
public String fetch(int queryId) {
return client.fetchQuery(queryId)
.map(response -> response.query())
.blockOptional()
@@ -20,4 +20,3 @@ public final class CloudQuerySource implements QuerySource {
"Cloud query was empty: " + queryId));
}
}

7
cygnus-onprem-app/src/main/java/matrix/nimble/query/QueryProvider.java
View File

@@ -2,10 +2,5 @@ package matrix.nimble.query;
public interface QueryProvider {
String getQuery(String queryId);
default String getQuery(int queryId) {
return getQuery("Query" + queryId);
}
String getQuery(int queryId);
}

3
cygnus-onprem-app/src/main/java/matrix/nimble/query/QuerySource.java
View File

@@ -3,6 +3,5 @@ package matrix.nimble.query;
@FunctionalInterface
public interface QuerySource {
String fetch(String queryId);
String fetch(int queryId);
}

22
cygnus-onprem-app/src/main/java/matrix/nimble/query/RedisCachingQueryProvider.java
View File

@@ -30,27 +30,27 @@ public final class RedisCachingQueryProvider implements QueryProvider {
}
@Override
public String getQuery(String queryId) {
String normalized = normalize(queryId);
Optional<String> cached = cached(normalized);
public String getQuery(int queryId) {
String cacheKey = cacheKey(queryId);
Optional<String> cached = cached(cacheKey);
if (cached.isPresent()) {
return cached.get();
}
synchronized (lock(normalized)) {
cached = cached(normalized);
synchronized (lock(cacheKey)) {
cached = cached(cacheKey);
if (cached.isPresent()) {
return cached.get();
}
try {
String query = cloudSource.fetch(normalized);
cache.put(normalized, cipher.encrypt(normalized, query), QUERY_TTL);
String query = cloudSource.fetch(queryId);
cache.put(cacheKey, cipher.encrypt(cacheKey, query), QUERY_TTL);
return query;
} catch (QueryProviderException exception) {
throw exception;
} catch (RuntimeException exception) {
throw new QueryProviderException(
"Unable to retrieve query: " + normalized, exception);
"Unable to retrieve query: " + queryId, exception);
}
}
}
@@ -72,10 +72,10 @@ public final class RedisCachingQueryProvider implements QueryProvider {
return locks[(queryId.hashCode() & Integer.MAX_VALUE) % locks.length];
}
private String normalize(String queryId) {
if (queryId == null || !queryId.matches("^[A-Za-z0-9._-]+$")) {
private String cacheKey(int queryId) {
if (queryId <= 0) {
throw new IllegalArgumentException("Invalid query ID");
}
return queryId;
return Integer.toString(queryId);
}
}

2
cygnus-onprem-app/src/test/java/matrix/nimble/FileFunctionsQueryCacheTest.java
View File

@@ -12,6 +12,6 @@ class FileFunctionsQueryCacheTest {
void delegatesLegacyQueryLookupToQueryProvider() throws Exception {
QueryProviders.install(queryId -> "provided:" + queryId);
FileFunctions files = new FileFunctions("TEST");
assertEquals("provided:Query3", files.GetQuery(3));
assertEquals("provided:3", files.GetQuery(3));
}
}

6
cygnus-onprem-app/src/test/java/matrix/nimble/query/AesGcmQueryCipherTest.java
View File

@@ -15,13 +15,13 @@ class AesGcmQueryCipherTest {
Arrays.fill(key, (byte) 7);
AesGcmQueryCipher cipher = new AesGcmQueryCipher(key);
String encrypted = cipher.encrypt("Query3", "select * from portfolio");
String encrypted = cipher.encrypt("3", "select * from portfolio");
assertNotEquals("select * from portfolio", encrypted);
assertTrue(encrypted.startsWith("v1."));
org.junit.jupiter.api.Assertions.assertEquals(
"select * from portfolio", cipher.decrypt("Query3", encrypted));
"select * from portfolio", cipher.decrypt("3", encrypted));
assertThrows(IllegalStateException.class,
() -> cipher.decrypt("Query4", encrypted));
() -> cipher.decrypt("4", encrypted));
}
}

8
cygnus-onprem-app/src/test/java/matrix/nimble/query/RedisCachingQueryProviderTest.java
View File

@@ -27,13 +27,13 @@ class RedisCachingQueryProviderTest {
},
cipher);
assertEquals("select!C0L!select 1", provider.getQuery("Query10"));
assertEquals("select!C0L!select 1", provider.getQuery("Query10"));
assertEquals("select!C0L!select 1", provider.getQuery(10));
assertEquals("select!C0L!select 1", provider.getQuery(10));
assertEquals(1, cloudCalls.get());
assertEquals(Duration.ofHours(3), cache.ttl.get());
org.junit.jupiter.api.Assertions.assertNotEquals(
"select!C0L!select 1", cache.values.get("Query10"));
"select!C0L!select 1", cache.values.get("10"));
}
@Test
@@ -55,7 +55,7 @@ class RedisCachingQueryProviderTest {
var executor = Executors.newFixedThreadPool(8);
try {
for (int index = 0; index < 20; index++) {
executor.submit(() -> provider.getQuery("Query20"));
executor.submit(() -> provider.getQuery(20));
}
} finally {
executor.shutdown();

1032
cygnus-onprem-app/src/main/resources/matrix/nimble/conf/nimble.qry → latest_queries.qry
View File

File diff suppressed because one or more lines are too long