<commit_before>

(* ---------------------------------------------------------------------------------------------
   The dbcache provides low-level access to the data in the db (both getting and setting).
   It is below the event-level: calling the functions in this module will not cause any events
   to be generated. For that reason, regular xapi code accessing the database must call in via
   the Db.* API (or, absolute worst-case (!) via the DB_ACCESS layer)

   The DBCache module contains 2 instances of the DB_CACHE module type: one instance for pool
   masters (that is an in-memory db-cache, with async write log pushed to sqlite db for persistent
   store); and one instance for pool slaves (that calls a marshalling protocol that retrieves/
   modifies directly from the ppol master).

   The DBCache module is also itself an instance of DB_CACHE: it's DB_CACHE fns are just a
   dispatcher (yes I know it's just a vtable and I could use objects ;) that calls Master/Slave
   depending on value in Pool_role.is_master ()
   --------------------------------------------------------------------------------------------- *)

(* Note: read_record could be improved - see comment below. It transfers data over network needlessly
   (although not much data really) -- main problem is it's somewhat confusing in its current
   form *)

open Db_lock
open Db_exn

module D=Debug.Debugger(struct let name="db_cache" end)
open D

type database_mode = Master | Slave

let database_mode : database_mode option ref = ref None 

exception Must_initialise_database_mode

(** Interface of DB_Cache implementations *)
module type DB_CACHE =
sig

  val dump_db_cache : Db_cache_types.db_dump_manifest -> Unix.file_descr -> unit

  val stats : unit -> (string * int) list

  val populate_cache : unit -> unit
  val populate_from  : Parse_db_conf.db_connection -> unit (* populate cache from specified file/format *)
  val spawn_db_flush_threads : unit -> unit

  (* Called by server to initialise db resources.
     Master implementation populates cache _and_ spawns db flush threads;
     Slave implementation sets up resources reqd for master connection + associated watchdog
  *)
  val initialise_db_cache : unit -> unit

  (* This will initialise a dbcache, but without syncing all db connections to the latest vsn;
     it's used by xapi-db-tool -- allowing it to read the xapi database without generating
     any writes to disk.. *)
  val initialise_db_cache_nosync : unit -> unit

  val display_sql_writelog : bool -> unit
  val flush_dirty : Parse_db_conf.db_connection -> bool
  val flush_and_exit : Parse_db_conf.db_connection -> int (* exit code *) -> unit

  val get_table_from_ref : string -> string option
  val is_valid_ref : string -> bool
  val read_refs : string -> string list
  val read_field_where : Db_cache_types.where_record -> string list
  val read_set_ref : Db_cache_types.where_record -> string list
  val create_row : Context.t -> string -> (string*string) list -> string -> unit
  val delete_row : Context.t -> string -> string -> unit
  val write_field : Context.t -> string -> string -> string -> string -> unit
  val read_field : Context.t -> string -> string -> string -> string
  val find_refs_with_filter : string -> Db_filter_types.expr -> string list

  val process_structured_field : Context.t -> (string*string) -> string -> string -> string -> Db_cache_types.structured_op_t -> unit
  val apply_delta_to_cache : Redo_log.t -> unit

  type db_record = (string * string) list * (string * (string list)) list (* dictionary of regular fields x dictionary of associated set_ref values *)
  val read_record : string -> string -> db_record
  val read_records_where : string -> Db_filter_types.expr -> (string * db_record) list
end

(** The dbcache that switches between master/slave accordingly *)
module DBCache : DB_CACHE =
struct
  type db_record = (string * string) list * (string * (string list)) list
      (* ---------------------------------------------------------------------------------------------------
	 Pool master implementation of DB_CACHE
	 --------------------------------------------------------------------------------------------------- *)

  (** An in-memory cache, used by pool master [not exposed beyond internals of DBCache] *)
  module Master : DB_CACHE =
  struct
    type db_record = (string * string) list * (string * (string list)) list
    module D = Debug.Debugger(struct let name = "sql" end)
    open D
    module W = Debug.Debugger(struct let name = "db_write" end)
    module R = Debug.Debugger(struct let name = "redo_log" end)
    
    open Db_cache_types
    open Db_action_helper
    open Db_backend

    let display_sql_writelog b = display_sql_writelog_val := b
		
    let redo_log_context_name = "redo_log"

    (* This fn is part of external interface, so need to take lock *)
    let get_table_from_ref objref =
      with_lock
	(fun () ->
	   try
	     Some (Hashtbl.find ref_table_map objref)
	   with Not_found -> None)

    let is_valid_ref objref =
      match (get_table_from_ref objref) with
	Some _ -> true
      | None -> false
	
    (** Return an association list of table name * record count *)
    let stats () = 
      with_lock 
	(fun () ->
	   Hashtbl.fold (fun name tbl acc ->
			   let size = Hashtbl.fold (fun _ _ acc -> acc + 1) tbl 0 in
			   (name, size) :: acc) cache [])

    let flush_dirty dbconn = Db_connections.flush_dirty_and_maybe_exit dbconn None
    let flush_and_exit dbconn ret_code = ignore (Db_connections.flush_dirty_and_maybe_exit dbconn (Some ret_code))
	
    (* Read field from cache *)
    let read_field context tblname fldname objref =
      with_lock
	(fun () ->
	   let row = find_row tblname objref in
	   lookup_field_in_row row fldname)

    let table_of_kvs kvs =
      let hashtbl = Hashtbl.create (List.length kvs) in
      List.iter (fun (k,v)->Hashtbl.replace hashtbl k v) kvs;
      hashtbl

    let save_in_redo_log context entry =
      if Redo_log.is_enabled() then begin
        Redo_log.write_delta (Generation.read_generation()) entry
          (fun () -> (* the function which will be invoked if a database write is required instead of a delta *)
            Backend_xml.flush_db_to_redo_log Db_backend.cache
          )
      end

	(** Finds the longest XML-compatible UTF-8 prefix of the given *)
	(** string, by truncating the string at the first incompatible *)
	(** character. Writes a warning to the debug log if truncation *)
	(** occurs.                                                    *)
	let ensure_utf8_xml string =
		let length = String.length string in
		let prefix = Encodings.UTF8_XML.longest_valid_prefix string in
		if length > String.length prefix then
			warn "string truncated to: '%s'." prefix;
		prefix

    (* Write field in cache *)
    let write_field context tblname objref fldname newval =
      with_lock
	(fun () ->
	   (* if uuid or reference then check uniqueness constraints: *)
	   if fldname=uuid_fname then
	     check_unique_table_constraints tblname (table_of_kvs [(uuid_fname, newval)])
	   else if fldname=reference_fname then
	     check_unique_table_constraints tblname (table_of_kvs [(reference_fname, newval)])
	   else if fldname=name_label_fname then
	     Ref_index.update_name_label objref newval;

	   let row = find_row tblname objref in
	   let current_val = lookup_field_in_row row fldname in

	   let other_tbl_refs = Eventgen.follow_references tblname in
	   let other_tbl_refs_for_this_field =
	     List.filter (fun (_,fld) -> fld=fldname) other_tbl_refs in

	   let newval = ensure_utf8_xml newval in

	   if current_val<>newval then
	     begin
	       W.debug "write_field %s,%s: %s |-> %s" tblname objref fldname newval;<commit_msg>CA-30784: When someone tries to modify an uuid of an API object, update also the Ref_index cache used by the CLI.<commit_after>

(* ---------------------------------------------------------------------------------------------
   The dbcache provides low-level access to the data in the db (both getting and setting).
   It is below the event-level: calling the functions in this module will not cause any events
   to be generated. For that reason, regular xapi code accessing the database must call in via
   the Db.* API (or, absolute worst-case (!) via the DB_ACCESS layer)

   The DBCache module contains 2 instances of the DB_CACHE module type: one instance for pool
   masters (that is an in-memory db-cache, with async write log pushed to sqlite db for persistent
   store); and one instance for pool slaves (that calls a marshalling protocol that retrieves/
   modifies directly from the ppol master).

   The DBCache module is also itself an instance of DB_CACHE: it's DB_CACHE fns are just a
   dispatcher (yes I know it's just a vtable and I could use objects ;) that calls Master/Slave
   depending on value in Pool_role.is_master ()
   --------------------------------------------------------------------------------------------- *)

(* Note: read_record could be improved - see comment below. It transfers data over network needlessly
   (although not much data really) -- main problem is it's somewhat confusing in its current
   form *)

open Db_lock
open Db_exn

module D=Debug.Debugger(struct let name="db_cache" end)
open D

type database_mode = Master | Slave

let database_mode : database_mode option ref = ref None 

exception Must_initialise_database_mode

(** Interface of DB_Cache implementations *)
module type DB_CACHE =
sig

  val dump_db_cache : Db_cache_types.db_dump_manifest -> Unix.file_descr -> unit

  val stats : unit -> (string * int) list

  val populate_cache : unit -> unit
  val populate_from  : Parse_db_conf.db_connection -> unit (* populate cache from specified file/format *)
  val spawn_db_flush_threads : unit -> unit

  (* Called by server to initialise db resources.
     Master implementation populates cache _and_ spawns db flush threads;
     Slave implementation sets up resources reqd for master connection + associated watchdog
  *)
  val initialise_db_cache : unit -> unit

  (* This will initialise a dbcache, but without syncing all db connections to the latest vsn;
     it's used by xapi-db-tool -- allowing it to read the xapi database without generating
     any writes to disk.. *)
  val initialise_db_cache_nosync : unit -> unit

  val display_sql_writelog : bool -> unit
  val flush_dirty : Parse_db_conf.db_connection -> bool
  val flush_and_exit : Parse_db_conf.db_connection -> int (* exit code *) -> unit

  val get_table_from_ref : string -> string option
  val is_valid_ref : string -> bool
  val read_refs : string -> string list
  val read_field_where : Db_cache_types.where_record -> string list
  val read_set_ref : Db_cache_types.where_record -> string list
  val create_row : Context.t -> string -> (string*string) list -> string -> unit
  val delete_row : Context.t -> string -> string -> unit
  val write_field : Context.t -> string -> string -> string -> string -> unit
  val read_field : Context.t -> string -> string -> string -> string
  val find_refs_with_filter : string -> Db_filter_types.expr -> string list

  val process_structured_field : Context.t -> (string*string) -> string -> string -> string -> Db_cache_types.structured_op_t -> unit
  val apply_delta_to_cache : Redo_log.t -> unit

  type db_record = (string * string) list * (string * (string list)) list (* dictionary of regular fields x dictionary of associated set_ref values *)
  val read_record : string -> string -> db_record
  val read_records_where : string -> Db_filter_types.expr -> (string * db_record) list
end

(** The dbcache that switches between master/slave accordingly *)
module DBCache : DB_CACHE =
struct
  type db_record = (string * string) list * (string * (string list)) list
      (* ---------------------------------------------------------------------------------------------------
	 Pool master implementation of DB_CACHE
	 --------------------------------------------------------------------------------------------------- *)

  (** An in-memory cache, used by pool master [not exposed beyond internals of DBCache] *)
  module Master : DB_CACHE =
  struct
    type db_record = (string * string) list * (string * (string list)) list
    module D = Debug.Debugger(struct let name = "sql" end)
    open D
    module W = Debug.Debugger(struct let name = "db_write" end)
    module R = Debug.Debugger(struct let name = "redo_log" end)
    
    open Db_cache_types
    open Db_action_helper
    open Db_backend

    let display_sql_writelog b = display_sql_writelog_val := b
		
    let redo_log_context_name = "redo_log"

    (* This fn is part of external interface, so need to take lock *)
    let get_table_from_ref objref =
      with_lock
	(fun () ->
	   try
	     Some (Hashtbl.find ref_table_map objref)
	   with Not_found -> None)

    let is_valid_ref objref =
      match (get_table_from_ref objref) with
	Some _ -> true
      | None -> false
	
    (** Return an association list of table name * record count *)
    let stats () = 
      with_lock 
	(fun () ->
	   Hashtbl.fold (fun name tbl acc ->
			   let size = Hashtbl.fold (fun _ _ acc -> acc + 1) tbl 0 in
			   (name, size) :: acc) cache [])

    let flush_dirty dbconn = Db_connections.flush_dirty_and_maybe_exit dbconn None
    let flush_and_exit dbconn ret_code = ignore (Db_connections.flush_dirty_and_maybe_exit dbconn (Some ret_code))
	
    (* Read field from cache *)
    let read_field context tblname fldname objref =
      with_lock
	(fun () ->
	   let row = find_row tblname objref in
	   lookup_field_in_row row fldname)

    let table_of_kvs kvs =
      let hashtbl = Hashtbl.create (List.length kvs) in
      List.iter (fun (k,v)->Hashtbl.replace hashtbl k v) kvs;
      hashtbl

    let save_in_redo_log context entry =
      if Redo_log.is_enabled() then begin
        Redo_log.write_delta (Generation.read_generation()) entry
          (fun () -> (* the function which will be invoked if a database write is required instead of a delta *)
            Backend_xml.flush_db_to_redo_log Db_backend.cache
          )
      end

	(** Finds the longest XML-compatible UTF-8 prefix of the given *)
	(** string, by truncating the string at the first incompatible *)
	(** character. Writes a warning to the debug log if truncation *)
	(** occurs.                                                    *)
	let ensure_utf8_xml string =
		let length = String.length string in
		let prefix = Encodings.UTF8_XML.longest_valid_prefix string in
		if length > String.length prefix then
			warn "string truncated to: '%s'." prefix;
		prefix

    (* Write field in cache *)
    let write_field context tblname objref fldname newval =
      with_lock
	(fun () ->
	   (* if uuid or reference then check uniqueness constraints: *)
	   if fldname=uuid_fname then begin
	     check_unique_table_constraints tblname (table_of_kvs [(uuid_fname, newval)]);
	     Ref_index.update_uuid objref newval;
	   end else if fldname=reference_fname then
	     check_unique_table_constraints tblname (table_of_kvs [(reference_fname, newval)])
	   else if fldname=name_label_fname then
	     Ref_index.update_name_label objref newval;

	   let row = find_row tblname objref in
	   let current_val = lookup_field_in_row row fldname in

	   let other_tbl_refs = Eventgen.follow_references tblname in
	   let other_tbl_refs_for_this_field =
	     List.filter (fun (_,fld) -> fld=fldname) other_tbl_refs in

	   let newval = ensure_utf8_xml newval in

	   if current_val<>newval then
	     begin
	       W.debug "write_field %s,%s: %s |-> %s" tblname objref fldname newval;
{'fraction_non_alphanumeric': 0.12048424289008455, 'fraction_numerical': 0.0012170125544452984, 'mean_word_length': 2.7006399620763215, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 125, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 6}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5186387', 'n_tokens_mistral': 4617, 'n_tokens_neox': 4349, 'n_words': 1630}